From c3c24b8c28ab1bd32b7a8e8f6c5be0e2378e3eea Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Fri, 16 Oct 2026 03:51:21 +0000 Subject: [PATCH 1/4] Queue io_uring submissions when SQ capacity is exhausted Defer accepted operations without reentrant completion dispatch, preserve cancellation and raw-fd lifetimes, and keep worker loops advancing pending work. Route listener shutdown to its owner worker and isolate submission and completion slow paths. Validate tiny-ring backpressure, cancellation, fd reuse, storage reads, and startup/shutdown races with the io_uring and runtime checks. --- CMakeLists.txt | 6 +- README.md | 1 + docs/architecture/io.md | 18 ++ docs/architecture/networking.md | 5 +- docs/architecture/runtime.md | 7 +- example/connect_timer_check.cpp | 108 +++++++ include/bycorf/io/io_uring_backend.h | 68 +++- include/bycorf/io/net_backend.h | 7 + include/bycorf/net/tcp_listener.h | 1 + include/bycorf/net/tcp_service.h | 16 +- include/bycorf/runtime/worker.h | 8 + src/io/dpdk_backend.cpp | 8 +- src/io/io_uring_backend.cpp | 436 +++++++++++++++++--------- src/net/tcp_listener.cpp | 4 + src/net/tcp_service.cpp | 65 ++-- src/net/tcp_stream.cpp | 22 +- src/runtime/worker.cpp | 19 +- tests/io_uring_backpressure_check.cpp | 392 +++++++++++++++++++++++ 18 files changed, 972 insertions(+), 219 deletions(-) create mode 100644 tests/io_uring_backpressure_check.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 7407857..34e51c6 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -426,6 +426,9 @@ if (BYCORF_BUILD_EXAMPLES OR BYCORF_BUILD_TESTS) add_executable(bycorf_tcp_buffered_read_check tests/tcp_buffered_read_check.cpp) target_link_libraries(bycorf_tcp_buffered_read_check PRIVATE bycorf::core) + add_executable(bycorf_io_uring_backpressure_check tests/io_uring_backpressure_check.cpp) + target_link_libraries(bycorf_io_uring_backpressure_check PRIVATE bycorf::core) + add_executable(bycorf_backend_selection_check example/backend_selection_check.cpp) target_link_libraries(bycorf_backend_selection_check PRIVATE bycorf::core) @@ -510,9 +513,10 @@ if (BYCORF_BUILD_TESTS) add_test(NAME bycorf_connect_timer_check COMMAND bycorf_connect_timer_check) add_test(NAME bycorf_connection_storage_check COMMAND bycorf_connection_storage_check) add_test(NAME bycorf_tcp_buffered_read_check COMMAND bycorf_tcp_buffered_read_check) + add_test(NAME bycorf_io_uring_backpressure_check COMMAND bycorf_io_uring_backpressure_check) add_test(NAME bycorf_backend_selection_check COMMAND bycorf_backend_selection_check) set_tests_properties(bycorf_connection_storage_check bycorf_backend_selection_check - bycorf_tcp_buffered_read_check + bycorf_tcp_buffered_read_check bycorf_io_uring_backpressure_check PROPERTIES TIMEOUT 90 LABELS "runtime;net;io_uring") set_tests_properties(bycorf_connect_timer_check PROPERTIES TIMEOUT 90 LABELS "runtime;net;io_uring" diff --git a/README.md b/README.md index 2163ffc..eb72198 100644 --- a/README.md +++ b/README.md @@ -100,6 +100,7 @@ a 30-second timeout; the other checks below have 90-second timeouts: | `bycorf_connect_timer_check` | Early shutdown, timers and cancellation, loopback TCP, refused connections, deadlines, address validation, and retirement of losing deadlines | | `bycorf_connection_storage_check` | Connection storage lifetime while suspended coroutines still borrow it | | `bycorf_tcp_buffered_read_check` | Bounded reads across queued receive buffers, partial buffers, unchanged ordinary reads, waiting, errors, and EOF | +| `bycorf_io_uring_backpressure_check` | Full submission queues, deferred sends and storage reads, queued cancellation, descriptor reuse, peer wakes, and shutdown | | `bycorf_backend_selection_check` | Unsupported backend requests, inactive optional backends, and immutable allocator/backend selection | Network checks use one unpinned worker and ephemeral loopback ports or Unix diff --git a/docs/architecture/io.md b/docs/architecture/io.md index 057afe1..714b015 100644 --- a/docs/architecture/io.md +++ b/docs/architecture/io.md @@ -38,6 +38,24 @@ submission/completion queues. Network-specific operations dispatch to the selected backend. The worker drives completion polling and batching, and does not create a second ring or worker pool for either accelerator. +Each io_uring backend owns a FIFO for accepted operations awaiting submission +space. Successful admission does not mean the kernel has received the request. +The backend retains its prepared descriptor; the caller retains the completion +tag, referenced buffers and parameter storage, and any registered resources +until the operation's terminal completion. Submission and completion polling +advance the queue on its owner worker, without dispatching completions inside +an operation's submission call. The event loop advances queued work before +parking, including when no earlier request can produce a completion. + +Cancellation preserves this ownership boundary: a queued operation retires +through deferred completion, while an issued operation remains live until its +terminal kernel completion. Cancelling a request does not itself release its +caller's state. Connection and listener close cancel their unsubmitted socket +operations before releasing the native descriptor, preventing deferred work +from reaching a reused descriptor. Other raw-descriptor callers retain their +descriptor until completion. Backend shutdown removes deferred work and +quiesces kernel access before the worker destroys suspended coroutine frames. + SPDK storage uses DMA buffers, NVMe namespaces and worker-owned queue pairs. Accepted commands that encounter temporary SPDK request-descriptor exhaustion remain in a bounded worker-local pool and wait on their queue pair's FIFO. diff --git a/docs/architecture/networking.md b/docs/architecture/networking.md index d47a0c1..7b1b32a 100644 --- a/docs/architecture/networking.md +++ b/docs/architecture/networking.md @@ -29,8 +29,9 @@ worker. Sessions that can suspend on non-socket work explicitly borrow connection storage before spawning and release it after their final users unwind. These holds delay storage reclamation, not transport closure, and must drain before the owning worker is destroyed. TCP service shutdown posts -listener-close work through each worker's foreign executor when the DPDK -backend is selected. +listener-close work through each worker's foreign executor for both backends. +The owner cancels unsubmitted kernel operations or enters its native TCP state +before releasing the listener; foreign threads never modify submission queues. Outbound numeric connects own their socket until connect and deadline completions have both retired. `ConnectTcpCancellable` additionally accepts a caller-owned, diff --git a/docs/architecture/runtime.md b/docs/architecture/runtime.md index a6b5062..c0fbede 100644 --- a/docs/architecture/runtime.md +++ b/docs/architecture/runtime.md @@ -57,9 +57,10 @@ 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. An unexpected backend wait failure -stops peer workers before entering those barriers and yields a nonzero exit. -Services outlive the server. +barriers so initialized peers can finish. Temporary I/O submission shortages +are retried by the worker loop. A permanent submission error or 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/example/connect_timer_check.cpp b/example/connect_timer_check.cpp index b5c8cfa..8d72c6b 100644 --- a/example/connect_timer_check.cpp +++ b/example/connect_timer_check.cpp @@ -18,10 +18,12 @@ // - bycorf::ConnectTcp (IORING_OP_CONNECT + deadline + loser cancel) // - bycorf::CancellableSleepFor (one-shot timer with a thread-safe cancel) // - bycorf::Runtime (stop requests racing worker startup) +// - bycorf::TcpService (foreign stop before/after listener startup) // Runs every check on one bycorf worker and exits non-zero on any failure. #include #include +#include #include #include @@ -37,6 +39,7 @@ #include "bycorf/io/storage.h" #include "bycorf/net/tcp_connect.h" #include "bycorf/net/tcp_listener.h" +#include "bycorf/net/tcp_service.h" #include "bycorf/net/tcp_stream.h" #include "bycorf/runtime/runtime.h" #include "bycorf/runtime/worker.h" @@ -95,6 +98,109 @@ void CheckStopDuringStartup(bool before_init) { : "startup stop: after Init, before Run"); } +class StopCheckService final : public bycorf::TcpService { + public: + explicit StopCheckService(std::uint16_t port) : TcpService(port) {} + + private: + bycorf::Task Serve(bycorf::TcpStream) override { + co_return absl::OkStatus(); + } +}; + +bycorf::Task RunServiceAndNotify( + StopCheckService& service, bycorf::Worker& worker, + bycorf::ServiceContext ctx, std::promise& finished) { + auto status = co_await service.Run(worker, ctx); + finished.set_value(status); + co_return status; +} + +bycorf::Task NotifyServiceStarted( + std::promise& started) { + started.set_value(absl::OkStatus()); + co_return absl::OkStatus(); +} + +void CheckTcpServiceStop(bool before_run) { + const int reservation = socket(AF_INET, SOCK_STREAM | SOCK_CLOEXEC, 0); + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + socklen_t length = sizeof(address); + const bool reserved = + reservation >= 0 && + bind(reservation, reinterpret_cast(&address), length) == 0 && + getsockname(reservation, reinterpret_cast(&address), + &length) == 0; + if (reservation >= 0) close(reservation); + Check(reserved, "TCP service stop: reserve loopback port"); + if (!reserved) return; + + StopCheckService service(ntohs(address.sin_port)); + service.Prepare(1); + bycorf::Runtime runtime; + std::promise started, finished; + auto ready = started.get_future(); + auto stopped = finished.get_future(); + std::promise release; + auto resume = release.get_future(); + runtime.Start( + 1, + [&](unsigned id, bycorf::Worker& worker) { + const auto status = + worker.Init({.ring_entries_ = 2, .recv_buffer_count_ = 0}); + if (!status.ok()) { + started.set_value(status); + return 1; + } + if (before_run) { + started.set_value(absl::OkStatus()); + resume.wait(); + } + const std::array hosts{"127.0.0.1"}; + bycorf::ServiceContext ctx{ + .bind_addresses_ = hosts, + .control_executor_ = runtime.GetForeignExecutor(id), + }; + worker.SpawnRoot(RunServiceAndNotify(service, worker, ctx, finished)); + // Ready-queue ordering announces startup only after Run has entered + // its accept wait. Stop then races an actual listener/SQ operation. + if (!before_run) worker.Spawn(NotifyServiceStarted(started)); + worker.Run(); + worker.Shutdown(); + worker.DestroyDetachedTasks(); + return 0; + }, + /*pin_workers=*/false); + const bool initialized = + ready.wait_for(5s) == std::future_status::ready && ready.get().ok(); + Check(initialized, "TCP service stop: startup reached gate"); + service.Stop(); + service.Stop(); + if (before_run) release.set_value(); + std::promise owner_marker; + auto owner_reached = owner_marker.get_future(); + if (before_run) { + // Keep the runtime alive until Run observes the earlier Stop. Forced + // runtime teardown must not hide a lost stop-before-publication race. + const bool completed = + stopped.wait_for(5s) == std::future_status::ready && stopped.get().ok(); + Check(completed, "TCP service stop: before Run publishes owner"); + } else { + // Stop posts listener closure before this marker on the same ingress. + // A consumed accept may still retain the socket until ring teardown, so + // require owner-side dispatch here rather than completion of that accept. + const bool posted = runtime.GetForeignExecutor(0).Notify( + [&owner_marker]() noexcept { owner_marker.set_value(); }); + Check(posted && owner_reached.wait_for(5s) == std::future_status::ready, + "TCP service stop: foreign close reaches owning worker"); + } + runtime.RequestStop(); + runtime.WaitUntilStopped(); + Check(runtime.exit_code() == 0, "TCP service stop: clean worker teardown"); +} + std::int64_t ElapsedMs(std::chrono::steady_clock::time_point start) { return std::chrono::duration_cast( std::chrono::steady_clock::now() - start) @@ -473,6 +579,8 @@ bycorf::Task RunAllChecks(bycorf::Worker& worker) { int main() { CheckStopDuringStartup(/*before_init=*/true); CheckStopDuringStartup(/*before_init=*/false); + CheckTcpServiceStop(/*before_run=*/true); + CheckTcpServiceStop(/*before_run=*/false); bycorf::Runtime runtime; runtime.Start( 1, diff --git a/include/bycorf/io/io_uring_backend.h b/include/bycorf/io/io_uring_backend.h index c1c4f31..be6df1d 100644 --- a/include/bycorf/io/io_uring_backend.h +++ b/include/bycorf/io/io_uring_backend.h @@ -20,8 +20,11 @@ #include #include +#include +#include #include #include +#include #include #include #include @@ -66,7 +69,11 @@ class IoUringBackend { return ring_.ring_fd; } // peers' MSG_RING target - // Typed submissions. On completion the backend invokes tag->Complete(worker, + // Success accepts an operation, including time waiting for SQ space. Callers + // retain tags, buffers and indirect arguments until terminal completion. + // Submitting never invokes a completion inline. Ordinary allocation failure + // is fatal, as for the worker's other queues. + // On completion the backend invokes tag->Complete(worker, // result, flags) — send (one-shot) and accept (multishot) both go through // this path; the backend is agnostic to which. recv does NOT use Complete: // the backend updates the Connection directly and resumes its reader. @@ -82,7 +89,8 @@ class IoUringBackend { // alive by embedding it in the awaitable. absl::Status SubmitConnect(int fd, const sockaddr* address, socklen_t address_length, IoCompletion* tag); - // Best-effort cancel of the in-flight SQE whose user_data is `target`. The + // Cancel a queued operation locally or best-effort cancel its in-flight SQE. + // Both paths deliver the target's terminal completion asynchronously. The // cancel request's own CQE is dropped (kCancelRequestTag); the target's CQE // still arrives — normally or with -ECANCELED — so the owner MUST wait for // it before reclaiming state the SQE referenced. -ENOENT (already done) is @@ -96,6 +104,12 @@ class IoUringBackend { absl::Status StartPeerDisconnectPoll(Connection* connection); absl::Status CancelPeerDisconnectPoll(Connection* connection); + // Retire unsubmitted kernel socket operations before closing/reusing a raw + // fd. Their cancellation completions are deferred to Poll; already consumed + // SQEs retain the kernel's file reference. Fixed-file operations are + // excluded. + void CancelUnsubmittedForFd(int fd) noexcept; + std::span ViewRecvBuffer(const Connection* connection, std::uint16_t buffer_id, std::size_t offset, @@ -127,6 +141,8 @@ class IoUringBackend { absl::Status Submit(); bool Poll(); // reap completions, dispatch, then re-arm recvs bool Wait(int timeout_ms); // block for a completion, dispatch, drain + // Software work or SQEs still requiring an enter must prevent worker sleep. + bool HasPending() const noexcept; // Cross-core: wake a peer worker that is parked (caller checks wake_seq). void WakeRemote(int peer_ring_fd) noexcept; @@ -134,6 +150,22 @@ class IoUringBackend { void WakeSelf() noexcept; private: + // Backend-owned submission state. Pointees remain caller-owned, just as for + // an SQE waiting in the real ring. Prepare the entire descriptor before + // publishing it so queue draining never observes a partially built request. + struct PendingIo { + explicit PendingIo(const io_uring_sqe& sqe) noexcept; + io_uring_sqe descriptor() const noexcept; + + // Linux's SQE contains a zero-length extension array, so store its object + // representation rather than embedding that non-standard member in C++. + std::array descriptor_; + }; + struct PendingCompletion { + std::uint64_t user_data_; + int result_; + }; + struct MultishotBufferRing { static constexpr std::uint16_t kGroupId = 1; io_uring_buf_ring* ring_ = nullptr; @@ -144,6 +176,16 @@ class IoUringBackend { }; io_uring_sqe* AcquireSqe(); + absl::Status SubmitOrQueue(const io_uring_sqe& prepared); + absl::Status QueuePrepared(const io_uring_sqe& prepared); + absl::Status SubmitSlow(); + bool PollDeferred(); + bool DrainPending(); + bool DrainPendingCompletions(); + template + bool CancelUnsubmittedMatching(Predicate matches, int result = -ECANCELED); + void RetireUnsubmittedCancels(std::uint64_t target); + absl::Status CancelByData(void* target, void* completion); bool InitMultishotRecv(); bool ArmWakePoll(); void HandleWakePoll(); @@ -151,22 +193,30 @@ class IoUringBackend { void HandlePeerDisconnect(Connection* connection, io_uring_cqe* cqe); void RecycleMultishotBuffer(std::uint16_t buffer_id); void DispatchCqe(io_uring_cqe* cqe); - bool - ReapCompletions(); // reap + dispatch only (no re-arm; used by AcquireSqe) + bool ReapCompletions(); void DrainRecvRearm(); io_uring ring_{}; + // Pack flags into the alignment gap before pointer-sized state. The rarely + // used ownership/registration flags then add no gaps between hot fields. bool initialized_ = false; - Worker* worker_ = nullptr; - IoBackendOptions options_{}; - MultishotBufferRing multishot_ring_{}; + bool may_have_unsubmitted_cancels_ = false; bool recv_multishot_enabled_ = false; - int wake_event_fd_ = -1; bool wake_poll_armed_ = false; bool owns_wake_fd_ = true; - std::vector recv_rearm_queue_; bool fixed_files_registered_ = false; bool buffers_registered_ = false; + Worker* worker_ = nullptr; + + // All three queues are checked by the event loop even when empty. Keep their + // metadata together ahead of receive-buffer storage and setup options. + std::deque pending_io_; + std::vector pending_completions_; + std::vector recv_rearm_queue_; + + MultishotBufferRing multishot_ring_{}; + int wake_event_fd_ = -1; + IoBackendOptions options_{}; }; } // namespace bycorf diff --git a/include/bycorf/io/net_backend.h b/include/bycorf/io/net_backend.h index 5c23ef9..915798c 100644 --- a/include/bycorf/io/net_backend.h +++ b/include/bycorf/io/net_backend.h @@ -84,6 +84,11 @@ class NetBackend { #endif return kernel_.SubmitCancel(target); } + // Cancel kernel socket operations that have not acquired their fd reference + // yet. Native DPDK handles must be excluded by the caller. + void CancelUnsubmittedForFd(int fd) noexcept { + kernel_.CancelUnsubmittedForFd(fd); + } absl::Status StartRecvMultishot(Connection* connection) { #if BYCORF_KERNEL_BYPASS if (dpdk_) return dpdk_->StartRecvMultishot(connection); @@ -179,6 +184,8 @@ class NetBackend { #endif return kernel_.Poll(); } + // The shared kernel ring can retain work even when DPDK serves networking. + bool HasPending() const noexcept { return kernel_.HasPending(); } bool Wait(int timeout_ms) { #if BYCORF_KERNEL_BYPASS if (dpdk_) return dpdk_->Wait(timeout_ms); diff --git a/include/bycorf/net/tcp_listener.h b/include/bycorf/net/tcp_listener.h index e9a7913..684a31d 100644 --- a/include/bycorf/net/tcp_listener.h +++ b/include/bycorf/net/tcp_listener.h @@ -116,6 +116,7 @@ class TcpListener { // recv operation is armed. AcceptUnregisteredAwaitable AcceptUnregistered() noexcept; Task> Accept(); + // Runs on the owning worker: closing also retires its unsubmitted accepts. absl::Status Close() noexcept; private: diff --git a/include/bycorf/net/tcp_service.h b/include/bycorf/net/tcp_service.h index e8855e3..8fb3bae 100644 --- a/include/bycorf/net/tcp_service.h +++ b/include/bycorf/net/tcp_service.h @@ -20,7 +20,6 @@ #include #include #include -#include #include #include "absl/status/statusor.h" @@ -80,7 +79,15 @@ class TcpService : public Service { struct WorkerListeners { std::vector values_; }; + struct WorkerControl { + ForeignExecutor executor_; + Worker* worker_ = nullptr; + // Both handles are written once before publication and stay immutable + // for the runtime lifetime. Stop never reads worker-owned listeners. + std::atomic ready_{false}; + }; + void CloseListeners(unsigned worker_id) noexcept; absl::Status StartSession(Worker& worker, Connection connection, std::shared_ptr tls); Task RunSession(Worker& worker, Connection* connection, @@ -95,11 +102,8 @@ class TcpService : public Service { unsigned thread_count_ = 0; std::vector session_workers_; std::atomic next_connection_worker_{0}; -#if BYCORF_KERNEL_BYPASS - std::mutex stop_mutex_; - bool stopping_ = false; - std::vector control_executors_; -#endif + std::atomic stopping_{false}; + std::unique_ptr controls_; // TODO: If long-lived connections develop uneven workloads after this // accept-time placement, add request-boundary live migration. It must first diff --git a/include/bycorf/runtime/worker.h b/include/bycorf/runtime/worker.h index 1e51abe..360193e 100644 --- a/include/bycorf/runtime/worker.h +++ b/include/bycorf/runtime/worker.h @@ -246,6 +246,11 @@ class Worker { absl::Status SubmitCancel(IoCompletion* target) { return backend_.SubmitCancel(target); } + // Before closing a kernel socket, retire accepted operations whose fd has + // not yet been consumed by the kernel. Completions remain deferred. + void CancelUnsubmittedForFd(int fd) noexcept { + backend_.CancelUnsubmittedForFd(fd); + } absl::Status EnsureRecvArmed(Connection* connection) { return backend_.StartRecvMultishot(connection); } @@ -432,6 +437,9 @@ class Worker { SpdkStorageBackend storage_backend_{}; #endif bool initialized_ = false; + // A fatal enter failure must leave RunOnce through the existing backend + // failure path, rather than spinning forever on accepted pending work. + bool submit_failed_ = false; std::atomic stop_requested_{false}; std::atomic stopping_{false}; WorkerOptions options_{}; diff --git a/src/io/dpdk_backend.cpp b/src/io/dpdk_backend.cpp index 3308043..ac9f2f5 100644 --- a/src/io/dpdk_backend.cpp +++ b/src/io/dpdk_backend.cpp @@ -946,9 +946,11 @@ bool DpdkBackend::Wait(int timeout_ms) { if (!Poll() && rte_ring_empty(lane.rx) && rte_ring_empty(lane.tx) && impl_->completions.empty()) { const auto submitted = kernel_.Submit(); - if (!submitted.ok()) - ok = false; - else { + if (!submitted.ok()) { + // Temporary enter backpressure needs another owner-worker round. Leave + // the interrupt cleanup below intact and avoid parking on deferred work. + ok = absl::IsUnavailable(submitted); + } else { const auto now = NowNs(), deadline = bycorf_bsd_deadline_ns(); if (deadline != UINT64_MAX) { const auto delay = deadline <= now diff --git a/src/io/io_uring_backend.cpp b/src/io/io_uring_backend.cpp index 82f09b4..fbbebb9 100644 --- a/src/io/io_uring_backend.cpp +++ b/src/io/io_uring_backend.cpp @@ -22,11 +22,13 @@ #include #include +#include #include #include #include #include +#include "absl/base/attributes.h" #include "bycorf/runtime/worker.h" #include "spdlog/spdlog.h" @@ -81,6 +83,16 @@ Connection* DecodePeerDisconnectConnection(void* data) { ~kConnectionTagMask); } +// Keep error classification and message allocation out of Submit's common +// success path, including when link-time optimization is enabled. +ABSL_ATTRIBUTE_COLD ABSL_ATTRIBUTE_NOINLINE absl::Status SubmitError(int rc) { + if (rc == -EBUSY || rc == -EINTR || rc == -EAGAIN) { + return absl::UnavailableError("io_uring submission temporarily blocked"); + } + return absl::InternalError("io_uring_submit_and_get_events failed: " + + std::string(std::strerror(-rc))); +} + } // namespace IoUringBackend::~IoUringBackend() { Shutdown(); } @@ -166,6 +178,14 @@ void IoUringBackend::Shutdown() { if (!initialized_) { return; } + // Teardown abandons suspended callers only after the ring is quiesced. Stop + // accepting work and drop software references before their frames can die; + // do not issue queued socket operations during shutdown. + initialized_ = false; + pending_io_.clear(); + pending_completions_.clear(); + may_have_unsubmitted_cancels_ = false; + recv_rearm_queue_.clear(); if (multishot_ring_.ring_ != nullptr) { io_uring_free_buf_ring(&ring_, multishot_ring_.ring_, multishot_ring_.entries_, @@ -232,157 +252,274 @@ absl::Status IoUringBackend::SubmitOpenDirect(std::string_view path, int flags, return absl::Status(absl::StatusCode::kFailedPrecondition, "fixed-file table is not registered"); } - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire open sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; io_uring_prep_openat_direct(sqe, AT_FDCWD, path.data(), flags, mode, file.index_); io_uring_sqe_set_data(sqe, tag); - return absl::OkStatus(); + return SubmitOrQueue(prepared); } absl::Status IoUringBackend::SubmitCloseDirect(FixedFile file, IoCompletion* tag) { - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire close sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; io_uring_prep_close_direct(sqe, file.index_); io_uring_sqe_set_data(sqe, tag); - return absl::OkStatus(); + return SubmitOrQueue(prepared); } absl::Status IoUringBackend::SubmitReadFixed(FixedFile file, FixedBuffer buffer, std::uint64_t offset, IoCompletion* tag) { - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire read sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; io_uring_prep_read_fixed(sqe, static_cast(file.index_), buffer.data_, static_cast(buffer.size_), offset, buffer.index_); sqe->flags |= IOSQE_FIXED_FILE; io_uring_sqe_set_data(sqe, tag); - return absl::OkStatus(); + return SubmitOrQueue(prepared); } absl::Status IoUringBackend::SubmitRead(FixedFile file, std::span buffer, std::uint64_t offset, IoCompletion* tag) { - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire read sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; io_uring_prep_read(sqe, static_cast(file.index_), buffer.data(), static_cast(buffer.size()), offset); sqe->flags |= IOSQE_FIXED_FILE; io_uring_sqe_set_data(sqe, tag); - return absl::OkStatus(); + return SubmitOrQueue(prepared); } absl::Status IoUringBackend::SubmitWrite(FixedFile file, std::span buffer, std::uint64_t offset, IoCompletion* tag) { - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire write sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; io_uring_prep_write(sqe, static_cast(file.index_), buffer.data(), static_cast(buffer.size()), offset); sqe->flags |= IOSQE_FIXED_FILE; io_uring_sqe_set_data(sqe, tag); - return absl::OkStatus(); + return SubmitOrQueue(prepared); } absl::Status IoUringBackend::SubmitWriteFixed(FixedFile file, FixedBuffer buffer, std::uint64_t offset, IoCompletion* tag) { - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire write sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; io_uring_prep_write_fixed(sqe, static_cast(file.index_), buffer.data_, static_cast(buffer.size_), offset, buffer.index_); sqe->flags |= IOSQE_FIXED_FILE; io_uring_sqe_set_data(sqe, tag); - return absl::OkStatus(); + return SubmitOrQueue(prepared); } absl::Status IoUringBackend::SubmitFdatasync(FixedFile file, IoCompletion* tag) { - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire fsync sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; io_uring_prep_fsync(sqe, static_cast(file.index_), IORING_FSYNC_DATASYNC); sqe->flags |= IOSQE_FIXED_FILE; io_uring_sqe_set_data(sqe, tag); - return absl::OkStatus(); + return SubmitOrQueue(prepared); } absl::Status IoUringBackend::SubmitTimeout(const __kernel_timespec& timeout, IoCompletion* tag) { - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire timeout sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; io_uring_prep_timeout(sqe, const_cast<__kernel_timespec*>(&timeout), 0, 0); io_uring_sqe_set_data(sqe, tag); - return absl::OkStatus(); + return SubmitOrQueue(prepared); } io_uring_sqe* IoUringBackend::AcquireSqe() { if (!initialized_) { return nullptr; } - io_uring_sqe* sqe = io_uring_get_sqe(&ring_); - if (sqe != nullptr) [[likely]] { - return sqe; - } - // SQ ring full (rare). Loop submit (frees SQ slots) + reap (frees CQ space) - // until an SQE is available — never returns null. We reap with - // ReapCompletions (NOT Poll): Poll would run DrainRecvRearm, whose - // StartRecvMultishot calls back into AcquireSqe and could recurse. Re-arm is - // deferred to Poll proper. - for (;;) { - io_uring_submit_and_get_events(&ring_); - sqe = io_uring_get_sqe(&ring_); - if (sqe != nullptr) { - return sqe; + return io_uring_get_sqe(&ring_); +} + +IoUringBackend::PendingIo::PendingIo(const io_uring_sqe& sqe) noexcept + : descriptor_(std::bit_cast(sqe)) {} + +io_uring_sqe IoUringBackend::PendingIo::descriptor() const noexcept { + return std::bit_cast(descriptor_); +} + +absl::Status IoUringBackend::SubmitOrQueue(const io_uring_sqe& prepared) { + // Older accepted work retains priority. Never submit/reap here: callers + // install their continuation and connection accounting after we return. + if (initialized_ && pending_io_.empty()) [[likely]] { + if (auto* sqe = AcquireSqe()) [[likely]] { + *sqe = prepared; + if (prepared.opcode == IORING_OP_ASYNC_CANCEL) [[unlikely]] { + may_have_unsubmitted_cancels_ = true; + } + return absl::OkStatus(); } - ReapCompletions(); } + return QueuePrepared(prepared); } -absl::Status IoUringBackend::Submit() { +// Deque growth and rejection diagnostics must not expand the ordinary SQE +// preparation path. Backlogs can be sustained, so this helper is not cold. +ABSL_ATTRIBUTE_NOINLINE absl::Status IoUringBackend::QueuePrepared( + const io_uring_sqe& prepared) { if (!initialized_) { - return absl::Status(absl::StatusCode::kFailedPrecondition, - "backend is not initialized"); + return absl::FailedPreconditionError("backend is not initialized"); } - const int rc = io_uring_submit_and_get_events(&ring_); - if (rc == -EBUSY) { - return absl::Status(absl::StatusCode::kUnavailable, - "io_uring completion ring busy"); + if (prepared.opcode == IORING_OP_ASYNC_CANCEL) { + may_have_unsubmitted_cancels_ = true; } - if (rc < 0) { - return absl::Status(absl::StatusCode::kInternal, - "io_uring_submit_and_get_events failed"); + pending_io_.push_back(PendingIo{prepared}); + return absl::OkStatus(); +} + +bool IoUringBackend::DrainPending() { + if (pending_io_.empty()) [[likely]] + return false; + bool prepared = false; + // At most one SQ's capacity is prepared per pass. Submissions and completion + // polling, rather than a producer-side busy loop, provide the next pass. + while (!pending_io_.empty()) { + auto* sqe = AcquireSqe(); + if (sqe == nullptr) break; + *sqe = pending_io_.front().descriptor(); + pending_io_.pop_front(); + prepared = true; + } + return prepared; +} + +bool IoUringBackend::HasPending() const noexcept { + return initialized_ && + (!pending_io_.empty() || !pending_completions_.empty() || + io_uring_sq_ready(&ring_) != 0); +} + +bool IoUringBackend::DrainPendingCompletions() { + if (pending_completions_.empty()) return false; + // A completion can submit/cancel other operations. Detach this round before + // dispatch, so new completions wait for the next event-loop pass. + std::vector completions; + completions.swap(pending_completions_); + for (const auto& completion : completions) { + io_uring_cqe cqe{}; + cqe.user_data = completion.user_data_; + cqe.res = completion.result_; + DispatchCqe(&cqe); + } + return true; +} + +template +bool IoUringBackend::CancelUnsubmittedMatching(Predicate matches, int result) { + if (!initialized_) return false; + bool cancelled = false; + auto retire = [&](const io_uring_sqe& sqe) { + pending_completions_.push_back({sqe.user_data, result}); + cancelled = true; + }; + // This ring is SINGLE_ISSUER without SQPOLL: the kernel only consumes SQEs + // during this worker's enter calls. Between calls, even a flushed but not + // consumed SQE is safe to neutralize. Include real SQEs, otherwise draining + // a pending socket request just before close would reintroduce fd reuse. + // We use neither SQE128 nor SQ_REWIND; liburing's SQ array is + // identity-mapped. + auto& sq = ring_.sq; + for (unsigned head = io_uring_load_sq_head(&ring_); head != sq.sqe_tail; + ++head) { + auto& sqe = sq.sqes[head & sq.ring_mask]; + if (!matches(sqe)) continue; + retire(sqe); + sqe = {}; + io_uring_prep_nop(&sqe); + io_uring_sqe_set_data(&sqe, &kCancelRequestTag); + } + for (auto it = pending_io_.begin(); it != pending_io_.end();) { + const auto sqe = it->descriptor(); + if (!matches(sqe)) { + ++it; + continue; + } + retire(sqe); + it = pending_io_.erase(it); } + return cancelled; +} + +absl::Status IoUringBackend::CancelByData(void* target, void* completion) { + if (!initialized_) { + return absl::FailedPreconditionError("backend is not initialized"); + } + const auto data = reinterpret_cast(target); + if (CancelUnsubmittedMatching( + [data](const io_uring_sqe& sqe) { return sqe.user_data == data; })) { + // No kernel cancel may outlive the synthetic target completion: its tag + // address could be reused by a new operation before that cancel is issued. + pending_completions_.push_back( + {reinterpret_cast(completion), 0}); + return absl::OkStatus(); + } + io_uring_sqe prepared{}; + io_uring_prep_cancel(&prepared, target, 0); + io_uring_sqe_set_data(&prepared, completion); + return SubmitOrQueue(prepared); +} + +void IoUringBackend::CancelUnsubmittedForFd(int fd) noexcept { + if (fd < 0) return; + CancelUnsubmittedMatching([fd](const io_uring_sqe& sqe) { + if ((sqe.flags & IOSQE_FIXED_FILE) != 0 || sqe.fd != fd) return false; + switch (sqe.opcode) { + case IORING_OP_SEND: + case IORING_OP_SENDMSG: + case IORING_OP_RECV: + case IORING_OP_ACCEPT: + case IORING_OP_CONNECT: + case IORING_OP_POLL_ADD: + return true; + default: + return false; + } + }); +} + +absl::Status IoUringBackend::Submit() { + // Enter does not dispatch application callbacks, so an empty software queue + // stays empty through this call and needs no before/after drain checks. + if (initialized_ && pending_io_.empty()) [[likely]] { + const int rc = io_uring_submit_and_get_events(&ring_); + if (rc >= 0) [[likely]] + return absl::OkStatus(); + return SubmitError(rc); + } + return SubmitSlow(); +} + +// Keep deque draining out of the common Submit instruction footprint even +// under LTO. Do not mark this cold: draining can be frequent under pressure. +ABSL_ATTRIBUTE_NOINLINE absl::Status IoUringBackend::SubmitSlow() { + if (!initialized_) { + return absl::FailedPreconditionError("backend is not initialized"); + } + DrainPending(); + const int rc = io_uring_submit_and_get_events(&ring_); + // Consuming SQ entries releases space before any operation completes. Prepare + // another bounded wave now; HasPending keeps the worker running to submit it. + DrainPending(); + if (rc < 0) [[unlikely]] + return SubmitError(rc); return absl::OkStatus(); } @@ -432,12 +569,10 @@ bool IoUringBackend::ArmWakePoll() { if (wake_event_fd_ < 0 || wake_poll_armed_) { return wake_event_fd_ >= 0; } - auto* sqe = AcquireSqe(); - if (sqe == nullptr) { - return false; - } - io_uring_prep_poll_add(sqe, wake_event_fd_, POLLIN); - io_uring_sqe_set_data(sqe, &kWakePollTag); + io_uring_sqe prepared{}; + io_uring_prep_poll_add(&prepared, wake_event_fd_, POLLIN); + io_uring_sqe_set_data(&prepared, &kWakePollTag); + if (!SubmitOrQueue(prepared).ok()) return false; wake_poll_armed_ = true; return true; } @@ -461,11 +596,8 @@ void IoUringBackend::HandleWakePoll() { absl::Status IoUringBackend::SubmitSend(const RegisteredFile& file, std::span buffer, IoCompletion* tag) { - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire send sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; io_uring_prep_send( sqe, file.is_fixed_ ? static_cast(file.fixed_index_) : file.fd_, buffer.data(), static_cast(buffer.size()), 0); @@ -473,17 +605,14 @@ absl::Status IoUringBackend::SubmitSend(const RegisteredFile& file, sqe->flags |= IOSQE_FIXED_FILE; } io_uring_sqe_set_data(sqe, tag); - return absl::OkStatus(); + return SubmitOrQueue(prepared); } absl::Status IoUringBackend::SubmitSendMsg(const RegisteredFile& file, const msghdr* message, IoCompletion* tag) { - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire sendmsg sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; io_uring_prep_sendmsg( sqe, file.is_fixed_ ? static_cast(file.fixed_index_) : file.fd_, message, 0); @@ -491,20 +620,17 @@ absl::Status IoUringBackend::SubmitSendMsg(const RegisteredFile& file, sqe->flags |= IOSQE_FIXED_FILE; } io_uring_sqe_set_data(sqe, tag); - return absl::OkStatus(); + return SubmitOrQueue(prepared); } absl::Status IoUringBackend::SubmitAcceptMultishot(int listen_fd, IoCompletion* tag) { - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire accept sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; io_uring_prep_multishot_accept(sqe, listen_fd, nullptr, nullptr, SOCK_NONBLOCK | SOCK_CLOEXEC); io_uring_sqe_set_data(sqe, tag); // plain IoCompletion* — same path as send - return absl::OkStatus(); + return SubmitOrQueue(prepared); } absl::Status IoUringBackend::SubmitConnect(int fd, const sockaddr* address, @@ -514,14 +640,11 @@ absl::Status IoUringBackend::SubmitConnect(int fd, const sockaddr* address, return absl::Status(absl::StatusCode::kInvalidArgument, "invalid connect request"); } - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire connect sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; io_uring_prep_connect(sqe, fd, address, address_length); io_uring_sqe_set_data(sqe, tag); - return absl::OkStatus(); + return SubmitOrQueue(prepared); } absl::Status IoUringBackend::SubmitCancel(IoCompletion* target) { @@ -529,14 +652,7 @@ absl::Status IoUringBackend::SubmitCancel(IoCompletion* target) { return absl::Status(absl::StatusCode::kInvalidArgument, "cancel target must not be null"); } - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire cancel sqe"); - } - io_uring_prep_cancel(sqe, target, 0); - io_uring_sqe_set_data(sqe, &kCancelRequestTag); - return absl::OkStatus(); + return CancelByData(target, &kCancelRequestTag); } absl::Status IoUringBackend::StartRecvMultishot(Connection* connection) { @@ -553,11 +669,8 @@ absl::Status IoUringBackend::StartRecvMultishot(Connection* connection) { if (!use_multishot && !connection->received_buffers_.empty()) { return absl::OkStatus(); } - auto* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire recv multishot sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; const int fd = connection->file_.is_fixed_ ? static_cast(connection->file_.fixed_index_) : connection->file_.fd_; @@ -579,6 +692,8 @@ absl::Status IoUringBackend::StartRecvMultishot(Connection* connection) { } io_uring_sqe_set_data(sqe, EncodeMultishotData(connection)); + auto status = SubmitOrQueue(prepared); + if (!status.ok()) return status; connection->recv_armed_ = true; connection->inflight_ops_ += 1; return absl::OkStatus(); @@ -590,17 +705,11 @@ absl::Status IoUringBackend::SubmitCancelRecv(Connection* connection, return absl::Status(absl::StatusCode::kInvalidArgument, "invalid recv cancel request"); } - auto* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::Status(absl::StatusCode::kUnavailable, - "failed to acquire recv cancel sqe"); - } - io_uring_prep_cancel(sqe, EncodeMultishotData(connection), 0); // Close needs only the receive's terminal CQE; a paused handoff also awaits // the cancellation request through its explicit completion tag. - io_uring_sqe_set_data( - sqe, tag != nullptr ? static_cast(tag) : &kCancelRequestTag); - return absl::OkStatus(); + return CancelByData( + EncodeMultishotData(connection), + tag != nullptr ? static_cast(tag) : &kCancelRequestTag); } absl::Status IoUringBackend::StartPeerDisconnectPoll(Connection* connection) { @@ -611,10 +720,8 @@ absl::Status IoUringBackend::StartPeerDisconnectPoll(Connection* connection) { connection->closing_) { return absl::OkStatus(); } - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::UnavailableError("failed to acquire peer disconnect poll sqe"); - } + io_uring_sqe prepared{}; + auto* sqe = &prepared; const int fd = connection->file_.is_fixed_ ? static_cast(connection->file_.fixed_index_) : connection->file_.fd_; @@ -622,6 +729,8 @@ absl::Status IoUringBackend::StartPeerDisconnectPoll(Connection* connection) { static_cast(POLLERR | POLLHUP | POLLRDHUP)); if (connection->file_.is_fixed_) sqe->flags |= IOSQE_FIXED_FILE; io_uring_sqe_set_data(sqe, EncodePeerDisconnectData(connection)); + auto status = SubmitOrQueue(prepared); + if (!status.ok()) return status; connection->peer_disconnect_poll_armed_ = true; connection->peer_disconnect_poll_cancel_requested_ = false; ++connection->inflight_ops_; @@ -633,13 +742,9 @@ absl::Status IoUringBackend::CancelPeerDisconnectPoll(Connection* connection) { connection->peer_disconnect_poll_cancel_requested_) { return absl::OkStatus(); } - io_uring_sqe* sqe = AcquireSqe(); - if (sqe == nullptr) { - return absl::UnavailableError( - "failed to acquire peer disconnect cancel sqe"); - } - io_uring_prep_cancel(sqe, EncodePeerDisconnectData(connection), 0); - io_uring_sqe_set_data(sqe, &kPeerDisconnectCancelTag); + auto status = CancelByData(EncodePeerDisconnectData(connection), + &kPeerDisconnectCancelTag); + if (!status.ok()) return status; connection->peer_disconnect_poll_cancel_requested_ = true; return absl::OkStatus(); } @@ -737,8 +842,8 @@ void IoUringBackend::HandleMultishotRecv(Connection* connection, connection->state_ == ConnectionState::kActive && !connection->closed_ && !connection->closing_ && !connection->recv_eof_) { - // Defer re-arm out of completion dispatch (DrainRecvRearm runs after the - // reap loop) so AcquireSqe is never reached from inside it. + // Re-arm after the completion batch so terminal receive accounting is + // settled before another receive is accepted into the submission queue. if (!connection->needs_recv_rearm_) { connection->needs_recv_rearm_ = true; recv_rearm_queue_.push_back(connection); @@ -771,7 +876,28 @@ void IoUringBackend::HandlePeerDisconnect(Connection* connection, } } +// A target may finish while its cancel still waits for SQ space. Retire that +// cancel before the callback can free/reuse its tag. Keep the scan and deque +// mutation outside the ordinary CQE dispatch instruction footprint. +ABSL_ATTRIBUTE_NOINLINE void IoUringBackend::RetireUnsubmittedCancels( + std::uint64_t target) { + bool other_cancels = false; + CancelUnsubmittedMatching( + [&](const io_uring_sqe& sqe) { + if (sqe.opcode != IORING_OP_ASYNC_CANCEL) return false; + if (sqe.addr == target) return true; + other_cancels = true; + return false; + }, + -ENOENT); + may_have_unsubmitted_cancels_ = other_cancels; +} + void IoUringBackend::DispatchCqe(io_uring_cqe* cqe) { + if (may_have_unsubmitted_cancels_ && (cqe->flags & IORING_CQE_F_MORE) == 0) + [[unlikely]] { + RetireUnsubmittedCancels(cqe->user_data); + } void* data = io_uring_cqe_get_data(cqe); if (data == &kWakePollTag) { HandleWakePoll(); @@ -834,25 +960,40 @@ void IoUringBackend::DrainRecvRearm() { } bool IoUringBackend::Poll() { - const bool processed = ReapCompletions(); - if (processed) { - DrainRecvRearm(); + if (!initialized_) return false; + bool processed = ReapCompletions(); + if (!pending_completions_.empty() || !recv_rearm_queue_.empty() || + !pending_io_.empty()) [[unlikely]] { + processed |= PollDeferred(); } return processed; } +// Poll's common path only checks queue metadata. Keep all deferred-work loops +// together without inlining them back into that path, including under LTO. +ABSL_ATTRIBUTE_NOINLINE bool IoUringBackend::PollDeferred() { + bool processed = DrainPendingCompletions(); + DrainRecvRearm(); + processed |= DrainPending(); + return processed; +} + absl::Status IoUringBackend::SubmitPoll(int fd, unsigned events, IoCompletion* tag) { if (fd < 0 || tag == nullptr) return absl::InvalidArgumentError("invalid poll request"); - auto* sqe = AcquireSqe(); - if (!sqe) return absl::UnavailableError("failed to acquire poll sqe"); + io_uring_sqe prepared{}; + auto* sqe = &prepared; io_uring_prep_poll_add(sqe, fd, events); io_uring_sqe_set_data(sqe, tag); - return absl::OkStatus(); + return SubmitOrQueue(prepared); } bool IoUringBackend::Wait(int timeout_ms) { + // A queued operation may be the very operation whose CQE would wake us. + // Return to the owner loop to enter/submit it before blocking for + // completions. + if (HasPending()) return true; io_uring_cqe* cqe = nullptr; int rc = 0; if (timeout_ms >= 0) { @@ -884,11 +1025,12 @@ bool IoUringBackend::Wait(int timeout_ms) { } void IoUringBackend::WakeRemote(int peer_ring_fd) noexcept { - io_uring_sqe* sqe = AcquireSqe(); // never null while initialized - io_uring_prep_msg_ring(sqe, peer_ring_fd, 0, + io_uring_sqe prepared{}; + io_uring_prep_msg_ring(&prepared, peer_ring_fd, 0, reinterpret_cast(&kCrossCoreWakeTag), 0); - io_uring_sqe_set_data(sqe, &kCrossCoreWakeTag); + io_uring_sqe_set_data(&prepared, &kCrossCoreWakeTag); + (void)SubmitOrQueue(prepared); } void IoUringBackend::WakeSelf() noexcept { diff --git a/src/net/tcp_listener.cpp b/src/net/tcp_listener.cpp index 8d0dd75..c57b69d 100644 --- a/src/net/tcp_listener.cpp +++ b/src/net/tcp_listener.cpp @@ -541,6 +541,10 @@ absl::Status TcpListener::Close() noexcept { fd_ = -1; accept_state_.CloseAllAcceptedFds(); absl::Status result = absl::OkStatus(); + if (fd >= 0 && worker_ != nullptr && !detail::IsDpdkSocket(fd)) { + // A queued accept must never bind to a later reuse of this descriptor. + worker_->CancelUnsubmittedForFd(fd); + } if (fd >= 0 && detail::CloseSocket(fd) != 0) { result = ErrnoToStatus(errno, "close listener failed"); } diff --git a/src/net/tcp_service.cpp b/src/net/tcp_service.cpp index fdd5e90..1736b95 100644 --- a/src/net/tcp_service.cpp +++ b/src/net/tcp_service.cpp @@ -55,10 +55,8 @@ void TcpService::Prepare(unsigned thread_count) { next_connection_worker_.store(0, std::memory_order_relaxed); listeners_.clear(); listeners_.resize(thread_count); -#if BYCORF_KERNEL_BYPASS - stopping_ = false; - control_executors_.resize(thread_count); -#endif + stopping_.store(false, std::memory_order_relaxed); + controls_ = std::make_unique(thread_count); } absl::Status TcpService::StartSession(Worker& worker, Connection connection, @@ -86,43 +84,42 @@ absl::Status TcpService::StartSession(Worker& worker, Connection connection, return absl::OkStatus(); } -void TcpService::Stop() noexcept { -#if BYCORF_KERNEL_BYPASS - if (DpdkNetworkEnabled()) { - std::lock_guard lock(stop_mutex_); - stopping_ = true; - // Native BSD socket operations must execute in their owner's VNET. The - // existing foreign ingress handles both idle-ring wakeup and publication. - for (unsigned i = 0; i < control_executors_.size(); ++i) { - if (control_executors_[i].valid()) { - (void)control_executors_[i].Notify([this, i]() noexcept { - for (auto& bound : listeners_[i].values_) - if (bound.listener_) bound.listener_->Close().IgnoreError(); - }); - } - } - return; +void TcpService::CloseListeners(unsigned worker_id) noexcept { + for (auto& bound : listeners_[worker_id].values_) { + if (bound.listener_) bound.listener_->Close().IgnoreError(); } -#endif - // Called from the Server's thread at shutdown; closing the listeners unblocks - // the accept loops (their multishot accept completes with -ECANCELED). - for (auto& worker : listeners_) { - for (auto& bound : worker.values_) { - if (bound.listener_) { - bound.listener_->Close().IgnoreError(); +} + +void TcpService::Stop() noexcept { + if (stopping_.exchange(true, std::memory_order_seq_cst)) return; + // Closing a kernel listener can cancel unsubmitted SQEs; BSD listeners need + // their owner's VNET. Both therefore use the owning worker's control ingress. + // Sequential consistency pairs with Run's publication/check: either Run sees + // stopping before binding, or this pass sees its published executor. + for (unsigned i = 0; i < thread_count_; ++i) { + if (controls_[i].ready_.load(std::memory_order_seq_cst)) { + if (!controls_[i].executor_.Notify( + [this, i]() noexcept { CloseListeners(i); })) { + // A live runtime can also reject the allocating notification on OOM. + // The allocation-free stop wake ensures this worker cannot continue + // serving indefinitely; Run's owner-side cleanup releases listeners. + controls_[i].worker_->RequestStop(); } } } } Task TcpService::Run(Worker& worker, ServiceContext ctx) { -#if BYCORF_KERNEL_BYPASS - { - std::lock_guard lock(stop_mutex_); - if (stopping_) co_return absl::OkStatus(); - control_executors_[worker.id()] = ctx.control_executor_; - } -#endif + WorkerControl& control = controls_[worker.id()]; + control.executor_ = ctx.control_executor_; + control.worker_ = &worker; + control.ready_.store(true, std::memory_order_seq_cst); + if (stopping_.load(std::memory_order_seq_cst)) co_return absl::OkStatus(); + // A stopping runtime can reject the close notification or leave its worker + // loop before delivering it. The frame is destroyed on its owning thread + // after I/O quiesces, so listener cleanup still has the required affinity. + auto close_listeners = absl::MakeCleanup( + [this, id = worker.id()]() noexcept { CloseListeners(id); }); WorkerListeners& owned = listeners_[worker.id()]; std::unordered_set seen; for (const Endpoint& endpoint : endpoints_) { diff --git a/src/net/tcp_stream.cpp b/src/net/tcp_stream.cpp index a59ae3a..9b3c14f 100644 --- a/src/net/tcp_stream.cpp +++ b/src/net/tcp_stream.cpp @@ -433,18 +433,17 @@ void ConnectOperation::CancelLoser(Worker& worker, IoCompletion* loser) { return; } loser_cancel_submitted_ = true; - // Best effort: if the cancel cannot be submitted (ring tearing down), the - // loser's own CQE still arrives — a timeout always fires — so awaiting both - // completions never deadlocks; the resume is merely late. + // Queued and issued targets both retire through a deferred completion. If + // teardown rejects cancellation, the worker quiesces I/O before abandoning + // this frame and its completion tags. (void)worker.SubmitCancel(loser); } void ConnectOperation::MaybeResume(Worker& worker) { - // Resuming requires both CQEs: the loser is cancelled asynchronously, and an - // early resume would let this frame be destroyed while the kernel still - // holds an SQE referencing it. AcquireSqe dispatches nested completions on - // its SQ-full retry, so a nested loser CQE can beat the outer winner handler - // here — resumed_ guards against the resulting double-enqueue. + // Both accepted operations must retire before this frame can be reclaimed, + // including a loser cancelled while still in the backend queue. Their + // completion paths converge here; resumed_ keeps parent notification + // one-shot. if (connect_done_ && timeout_done_ && !resumed_) { resumed_ = true; worker.Enqueue(awaiting_); @@ -718,10 +717,9 @@ void TcpConnectCancellation::Cancel() noexcept { if (operation_ != nullptr) (void)worker_->SubmitCancel(operation_); } -// Register cancellation only after the existing awaitable has submitted its -// connect and timer. Submission may dispatch nested completions while making -// SQ space: an earlier Cancel records intent and is replayed after binding. -// The existing awaitable still owns retirement of both completion tags. +// Bind cancellation only after the awaitable accepts its connect and optional +// timer. Cancellation recorded before binding is replayed against the accepted +// connect; the awaitable retains both completion tags until they retire. class CancellableConnectAwaitable { public: CancellableConnectAwaitable(Worker& worker, int fd, const sockaddr* address, diff --git a/src/runtime/worker.cpp b/src/runtime/worker.cpp index 178e0f4..2d6fc2f 100644 --- a/src/runtime/worker.cpp +++ b/src/runtime/worker.cpp @@ -221,6 +221,7 @@ absl::Status Worker::Init(const WorkerOptions& options) { // Advertise our ring so other workers can wake us via MSG_RING. cross_core_->mailbox(id_).ring_fd_ = backend_.WakeHandle(); + submit_failed_ = false; initialized_ = true; return absl::OkStatus(); } @@ -373,6 +374,11 @@ void Worker::BeginClose(Connection* connection, absl::Status reason, if (connection->file_.fd_ >= 0) { const int fd = connection->file_.fd_; + if (!detail::IsDpdkSocket(fd)) { + // An accepted operation may still live in the software queue or SQ. + // Retire it before close lets another socket reuse this descriptor. + backend_.CancelUnsubmittedForFd(fd); + } connection->file_.fd_ = -1; connection->closed_ = true; detail::CloseSocket(fd); @@ -1059,12 +1065,16 @@ void Worker::Flush() { FlushWakes(); const auto status = backend_.Submit(); if (!status.ok()) [[unlikely]] { + // CQ backpressure and interrupted enters are retried by the next round. + // A permanent error cannot make progress by waiting for queue capacity. + if (absl::IsUnavailable(status)) return; spdlog::error("worker submit failed: {}", status.message()); + submit_failed_ = true; } } bool Worker::RunOnce(bool wait_for_completion) { - if (!initialized_) [[unlikely]] { + if (!initialized_ || submit_failed_) [[unlikely]] { return false; } @@ -1133,6 +1143,7 @@ bool Worker::RunOnce(bool wait_for_completion) { // maintenance slice. Peers can then make progress while this worker runs // background CPU work, instead of waiting an extra slice for every reply. Flush(); + if (submit_failed_) return false; const std::int64_t background_start = CycleNow(); const std::size_t background_resumes = DrainBackgroundUntil(background_start + static_cast( @@ -1156,6 +1167,10 @@ bool Worker::RunOnce(bool wait_for_completion) { // parked are resumed by the next iteration, which always reaches here // before it can block. Flush(); + if (submit_failed_) return false; + // Submission can free SQ slots without producing a completion. Keep both + // normal rounds and shutdown drains running until deferred work advances. + did_work |= backend_.HasPending(); ReclaimConnections(); MaybeResetRuntimeWindow(); RecordRound(round_start); @@ -1168,7 +1183,7 @@ bool Worker::RunOnce(bool wait_for_completion) { } if (BusyPoll()) { - return true; + return !submit_failed_; } #if BYCORF_KERNEL_BYPASS diff --git a/tests/io_uring_backpressure_check.cpp b/tests/io_uring_backpressure_check.cpp new file mode 100644 index 0000000..72b8cb8 --- /dev/null +++ b/tests/io_uring_backpressure_check.cpp @@ -0,0 +1,392 @@ +/* + * 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 + +#include "bycorf/io/io_uring_backend.h" +#include "bycorf/runtime/worker.h" + +namespace { + +#define CHECK(condition) \ + do { \ + if (!(condition)) { \ + std::cerr << "FAIL line " << __LINE__ << ": " #condition << '\n'; \ + std::exit(1); \ + } \ + } while (false) + +struct Completion final : bycorf::IoCompletion { + unsigned calls = 0; + int result = 0; + + void Complete(bycorf::Worker&, int value, unsigned flags) override { + CHECK(++calls == 1); + CHECK(flags == bycorf::kCompletionNone); + result = value; + } +}; + +struct Fixture { + Fixture() { + const auto status = backend.Init( + {.ring_entries_ = 2, .recv_buffer_count_ = 0}, &worker, -1); + if (!status.ok()) std::cerr << status << '\n'; + CHECK(status.ok()); + } + + bycorf::Worker worker; + bycorf::IoUringBackend backend; +}; + +template +void Drain(bycorf::IoUringBackend& backend, Predicate done) { + const auto deadline = + std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (!done()) { + CHECK(std::chrono::steady_clock::now() < deadline); + const auto status = backend.Submit(); + CHECK(status.ok() || absl::IsUnavailable(status)); + backend.Poll(); + } +} + +void CheckSendsAndWake() { + Fixture source, peer; + int sockets[2]; + CHECK(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, + sockets) == 0); + constexpr std::size_t kCount = 64; + std::array tags; + std::array payload; + for (std::size_t i = 0; i < kCount; ++i) { + payload[i] = std::byte(i); + CHECK(source.backend + .SubmitSend({.fd_ = sockets[0]}, std::span(payload).subspan(i, 1), + &tags[i]) + .ok()); + } + source.backend.WakeRemote(peer.backend.WakeHandle()); + for (const auto& tag : tags) CHECK(tag.calls == 0); + + // A completion-only check misses hidden io_uring_enter calls that free SQ + // slots without reaping. The peer must see no data before the event loop + // explicitly submits this burst, even though it exceeds the ring capacity. + std::array received; + CHECK(recv(sockets[1], received.data(), received.size(), MSG_DONTWAIT) == -1); + CHECK(errno == EAGAIN || errno == EWOULDBLOCK); + CHECK(peer.backend.Submit().ok()); + CHECK(!peer.backend.Poll()); + + bool peer_woken = false; + Drain(source.backend, [&] { + peer_woken |= peer.backend.Poll(); + for (const auto& tag : tags) + if (tag.calls == 0) return false; + return peer_woken; + }); + for (const auto& tag : tags) CHECK(tag.result == 1); + CHECK(recv(sockets[1], received.data(), received.size(), MSG_DONTWAIT) == + static_cast(received.size())); + CHECK(received == payload); + CHECK(close(sockets[0]) == 0); + CHECK(close(sockets[1]) == 0); + std::cout << "PASS deferred sends, FIFO payloads and queued peer wake\n"; +} + +void CheckQueuedCancellation() { + Fixture fixture; + const int event = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); + CHECK(event >= 0); + std::array tags; + for (std::size_t i = 0; i + 1 < tags.size(); ++i) + CHECK(fixture.backend.SubmitPoll(event, POLLIN, &tags[i]).ok()); + const __kernel_timespec timeout{.tv_sec = 60, .tv_nsec = 0}; + CHECK(fixture.backend.SubmitTimeout(timeout, &tags.back()).ok()); + // Both the target and its cancel are behind a full SQ. Cancellation must + // still deliver the target's terminal completion before its tag is released. + CHECK(fixture.backend.SubmitCancel(&tags.back()).ok()); + for (const auto& tag : tags) CHECK(tag.calls == 0); + Drain(fixture.backend, [&] { return tags.back().calls != 0; }); + CHECK(tags.back().result == -ECANCELED); + for (std::size_t i = 0; i + 1 < tags.size(); ++i) CHECK(tags[i].calls == 0); + + const std::uint64_t one = 1; + CHECK(write(event, &one, sizeof(one)) == sizeof(one)); + Drain(fixture.backend, [&] { + for (const auto& tag : tags) + if (tag.calls == 0) return false; + return true; + }); + for (std::size_t i = 0; i + 1 < tags.size(); ++i) + CHECK((tags[i].result & POLLIN) != 0); + CHECK(fixture.backend.Submit().ok()); + fixture.backend.Poll(); + CHECK(tags.back().calls == 1); + CHECK(close(event) == 0); + std::cout << "PASS queued cancellation and exactly-once completions\n"; +} + +void CheckClosedDescriptorReuse() { + Fixture fixture; + int original[2], replacement[2]; + CHECK(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, + original) == 0); + CHECK(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, + replacement) == 0); + const std::array payload{std::byte{0x5a}}; + std::array tags; + // Initialization occupies one SQ slot; the first send is in the real SQ, + // while later sends are software-queued. Neither may outlive this fd's + // identity, and cancellation callbacks must remain deferred. + for (auto& tag : tags) + CHECK(fixture.backend.SubmitSend({.fd_ = original[0]}, payload, &tag).ok()); + fixture.backend.CancelUnsubmittedForFd(original[0]); + for (const auto& tag : tags) CHECK(tag.calls == 0); + CHECK(close(original[0]) == 0); + CHECK(dup2(replacement[0], original[0]) == original[0]); + CHECK(close(replacement[0]) == 0); + + Drain(fixture.backend, [&] { + for (const auto& tag : tags) + if (tag.calls == 0) return false; + return true; + }); + for (const auto& tag : tags) CHECK(tag.result == -ECANCELED); + std::byte received{}; + CHECK(recv(original[1], &received, 1, MSG_DONTWAIT) == 0); + CHECK(recv(replacement[1], &received, 1, MSG_DONTWAIT) == -1); + CHECK(errno == EAGAIN || errno == EWOULDBLOCK); + + Completion fresh; + CHECK(fixture.backend.SubmitSend({.fd_ = original[0]}, payload, &fresh).ok()); + Drain(fixture.backend, [&] { return fresh.calls != 0; }); + CHECK(fresh.result == 1); + CHECK(recv(replacement[1], &received, 1, MSG_DONTWAIT) == 1); + CHECK(received == payload.front()); + CHECK(close(original[0]) == 0); + CHECK(close(original[1]) == 0); + CHECK(close(replacement[1]) == 0); + std::cout << "PASS cancellation before close protects reused descriptors\n"; +} + +void CheckQueuedReceiveCancellation() { + Fixture fixture; + const int event = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); + CHECK(event >= 0); + std::array blockers; + for (auto& tag : blockers) + CHECK(fixture.backend.SubmitPoll(event, POLLIN, &tag).ok()); + + int sockets[2]; + CHECK(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, + sockets) == 0); + bycorf::Connection connection; + connection.worker_ = &fixture.worker; + connection.file_.fd_ = sockets[0]; + connection.closed_ = false; + CHECK(send(sockets[1], "x", 1, MSG_NOSIGNAL) == 1); + CHECK(fixture.backend.StartRecvMultishot(&connection).ok()); + CHECK(connection.recv_armed_ && connection.inflight_ops_ == 1); + connection.recv_paused_ = true; + Completion cancelled; + CHECK(fixture.backend.SubmitCancelRecv(&connection, &cancelled).ok()); + CHECK(cancelled.calls == 0); + CHECK(connection.recv_armed_ && connection.inflight_ops_ == 1); + Drain(fixture.backend, [&] { return cancelled.calls != 0; }); + CHECK(cancelled.result == 0); + CHECK(!connection.recv_armed_ && connection.inflight_ops_ == 0); + CHECK(connection.received_buffers_.empty()); + char received = 0; + CHECK(recv(sockets[0], &received, 1, MSG_DONTWAIT) == 1); + CHECK(received == 'x'); + + const std::uint64_t one = 1; + CHECK(write(event, &one, sizeof(one)) == sizeof(one)); + Drain(fixture.backend, [&] { + for (const auto& tag : blockers) + if (tag.calls == 0) return false; + return true; + }); + CHECK(close(event) == 0); + CHECK(close(sockets[0]) == 0); + CHECK(close(sockets[1]) == 0); + std::cout << "PASS queued receive cancellation drains connection ownership\n"; +} + +void CheckLateCancelBeforeTagReuse() { + Fixture fixture; + const int event = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); + CHECK(event >= 0); + int sockets[2]; + CHECK(socketpair(AF_UNIX, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0, + sockets) == 0); + bycorf::Connection connection; + connection.worker_ = &fixture.worker; + connection.file_.fd_ = sockets[0]; + connection.closed_ = false; + CHECK(fixture.backend.StartRecvMultishot(&connection).ok()); + CHECK(fixture.backend.Submit().ok()); + CHECK(!fixture.backend.HasPending()); + + std::array blockers; + for (auto& tag : blockers) + CHECK(fixture.backend.SubmitPoll(event, POLLIN, &tag).ok()); + Completion cancelled; + CHECK(fixture.backend.SubmitCancelRecv(&connection, &cancelled).ok()); + CHECK(cancelled.calls == 0); + CHECK(send(sockets[1], "x", 1, MSG_NOSIGNAL) == 1); + // Run deferred kernel completion work without publishing or submitting the + // full SQ/backlog. The receive therefore finishes naturally while its cancel + // is still wholly owned by the backend's submission queue. + alarm(2); + CHECK(io_uring_enter(fixture.backend.WakeHandle(), 0, 1, + IORING_ENTER_GETEVENTS, nullptr) >= 0); + alarm(0); + CHECK(fixture.backend.Poll()); + CHECK(!connection.recv_armed_ && connection.inflight_ops_ == 0); + CHECK(connection.received_buffers_.size() == 1); + CHECK(connection.read_buffer_.front() == std::byte{'x'}); + // This must be delivered by the same Poll, before any queued cancel can + // enter the kernel. Merely checking the next read would allow FIFO submission + // to hide a stale cancel that happens to run before the reused tag is armed. + CHECK(cancelled.calls == 1 && cancelled.result == -ENOENT); + + connection.received_buffers_.clear(); + CHECK(fixture.backend.StartRecvMultishot(&connection).ok()); + CHECK(connection.recv_armed_ && connection.inflight_ops_ == 1); + CHECK(send(sockets[1], "y", 1, MSG_NOSIGNAL) == 1); + const std::uint64_t one = 1; + CHECK(write(event, &one, sizeof(one)) == sizeof(one)); + Drain(fixture.backend, [&] { + for (const auto& tag : blockers) + if (tag.calls == 0) return false; + return !connection.received_buffers_.empty(); + }); + CHECK(cancelled.calls == 1); + CHECK(!connection.recv_armed_ && connection.inflight_ops_ == 0); + CHECK(connection.read_buffer_.front() == std::byte{'y'}); + CHECK(close(event) == 0); + CHECK(close(sockets[0]) == 0); + CHECK(close(sockets[1]) == 0); + std::cout + << "PASS natural completion retires queued cancel before tag reuse\n"; +} + +void CheckPendingWaitAndShutdown() { + Fixture fixture; + const int event = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); + CHECK(event >= 0); + { + std::array tags; + for (auto& tag : tags) + CHECK(fixture.backend.SubmitPoll(event, POLLIN, &tag).ok()); + CHECK(fixture.backend.SubmitCancel(&tags.back()).ok()); + CHECK(fixture.backend.HasPending()); + // None of these polls can complete. Blocking here would strand both the + // unsubmitted work and the software cancellation completion behind it. + alarm(2); + CHECK(fixture.backend.Wait(-1)); + alarm(0); + for (const auto& tag : tags) CHECK(tag.calls == 0); + fixture.backend.Shutdown(); + CHECK(!fixture.backend.HasPending()); + CHECK(!fixture.backend.Poll()); + } + CHECK(close(event) == 0); + + // Old tags are now destroyed. Reusing the backend must not issue abandoned + // SQEs or dispatch their synthetic completions into the old caller's state. + CHECK(fixture.backend + .Init({.ring_entries_ = 2, .recv_buffer_count_ = 0}, + &fixture.worker, -1) + .ok()); + Completion fresh; + const __kernel_timespec timeout{.tv_sec = 0, .tv_nsec = 1}; + CHECK(fixture.backend.SubmitTimeout(timeout, &fresh).ok()); + Drain(fixture.backend, [&] { return fresh.calls != 0; }); + CHECK(fresh.result == -ETIME); + std::cout << "PASS pending work prevents park and shutdown drops old state\n"; +} + +void CheckStorageReads() { + Fixture fixture; + char path[] = "/tmp/bycorf-uring-backpressure-XXXXXX"; + const int file = mkstemp(path); + CHECK(file >= 0); + std::array payload; + for (std::size_t i = 0; i < payload.size(); ++i) payload[i] = std::byte(i); + CHECK(write(file, payload.data(), payload.size()) == + static_cast(payload.size())); + CHECK(fixture.backend.RegisterFixedFiles(1).ok()); + Completion opened; + CHECK(fixture.backend.SubmitOpenDirect(path, O_RDONLY, 0, {0}, &opened).ok()); + Drain(fixture.backend, [&] { return opened.calls != 0; }); + CHECK(opened.result == 0); + CHECK(unlink(path) == 0); + CHECK(close(file) == 0); + + std::array tags; + std::array received{}; + const iovec region{.iov_base = received.data(), .iov_len = received.size()}; + CHECK(fixture.backend.RegisterBuffers(std::span(®ion, 1)).ok()); + for (std::size_t i = 0; i < tags.size(); ++i) { + const auto buffer = std::span(received).subspan(i * 4, 4); + CHECK((i % 2 == 0 + ? fixture.backend.SubmitRead({0}, buffer, i * 4, &tags[i]) + : fixture.backend.SubmitReadFixed( + {0}, {.data_ = buffer.data(), .size_ = buffer.size()}, + i * 4, &tags[i])) + .ok()); + } + for (const auto& tag : tags) CHECK(tag.calls == 0); + Drain(fixture.backend, [&] { + for (const auto& tag : tags) + if (tag.calls == 0) return false; + return true; + }); + for (const auto& tag : tags) CHECK(tag.result == 4); + CHECK(received == payload); + Completion closed; + CHECK(fixture.backend.SubmitCloseDirect({0}, &closed).ok()); + Drain(fixture.backend, [&] { return closed.calls != 0; }); + CHECK(closed.result == 0); + std::cout << "PASS queued fixed-file reads preserve buffers and offsets\n"; +} + +} // namespace + +int main() { + CheckSendsAndWake(); + CheckQueuedCancellation(); + CheckClosedDescriptorReuse(); + CheckQueuedReceiveCancellation(); + CheckLateCancelBeforeTagReuse(); + CheckPendingWaitAndShutdown(); + CheckStorageReads(); + return 0; +} From e09bd7db8974d5d0d31ae2b0269197b51a903791 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Fri, 16 Oct 2026 04:48:13 +0000 Subject: [PATCH 2/4] Queue SPDK I/O without a bounded callback pool --- README.md | 6 +- docs/architecture/io.md | 18 +- include/bycorf/io/completion.h | 13 +- include/bycorf/io/spdk_storage.h | 31 ++- include/bycorf/runtime/worker.h | 3 + src/io/spdk_storage.cpp | 241 ++++++++++--------- tests/spdk_backpressure_stub_check.cpp | 320 ++++++++++++++++++++++--- 7 files changed, 455 insertions(+), 177 deletions(-) diff --git a/README.md b/README.md index eb72198..00f10b1 100644 --- a/README.md +++ b/README.md @@ -114,9 +114,9 @@ With `BYCORF_KERNEL_BYPASS=ON`, CTest also runs the FreeBSD local-connect regression and `bycorf_spdk_backpressure_stub_check`. The SPDK check uses the real backend and Worker with link-time stubs for EAL initialization and NVMe device calls. It covers queue exhaustion, FIFO retries, independent qpairs, -submission/completion errors, request-pool reuse, and shutdown draining without -NVMe hardware, hugepages, or device binding. It still requires working io_uring -and the normal runtime memlock allowance. Run it alone with: +large backlogs, submission/completion errors, completion-tag reuse, and shutdown +draining without NVMe hardware, hugepages, or device binding. It still requires +working io_uring and the normal runtime memlock allowance. Run it alone with: ```bash ctest --test-dir build -R '^bycorf_spdk_backpressure_stub_check$' --output-on-failure diff --git a/docs/architecture/io.md b/docs/architecture/io.md index 714b015..58f0b95 100644 --- a/docs/architecture/io.md +++ b/docs/architecture/io.md @@ -58,13 +58,17 @@ quiesces kernel access before the worker destroys suspended coroutine frames. SPDK storage uses DMA buffers, NVMe namespaces and worker-owned queue pairs. Accepted commands that encounter temporary SPDK request-descriptor exhaustion -remain in a bounded worker-local pool and wait on their queue pair's FIFO. -Completion polling retries them; their buffers remain owned by the awaiting -caller until completion. Polling and shutdown drains retain the owning -worker's thread-local context for completion dispatch. Deferred commands -participate in outstanding-I/O accounting, preventing premature sleep, close -or shutdown. Submission errors that cannot make progress through a completion -are reported to the caller. +wait on their queue pair's FIFO, whose storage grows independently of other +queue pairs' capacity. Completion polling retries them; the backend releases +its saved submission parameters once SPDK accepts the command. SPDK returns +the caller's completion tag directly through its callback context. Each +unfinished storage operation has a distinct tag, which retains its expected +successful result; the caller keeps the tag and buffers alive until completion. +The completion callback may reuse or destroy the tag. Polling and shutdown +drains retain the owning worker's thread-local context for completion dispatch. +Deferred commands participate in outstanding-I/O accounting, preventing +premature sleep, close or shutdown. Submission errors that cannot make progress +through a completion are reported to the caller. The io_uring selection uses ordinary aligned buffers and kernel file/block I/O. SPDK completion polling is inactive when storage uses io_uring; in-flight SPDK storage prevents sleeping before its completions can be polled. As with the diff --git a/include/bycorf/io/completion.h b/include/bycorf/io/completion.h index 2ae9381..0a2f91d 100644 --- a/include/bycorf/io/completion.h +++ b/include/bycorf/io/completion.h @@ -34,7 +34,10 @@ enum CompletionFlags : unsigned { // Backend-neutral completion callback. The io backend invokes Complete() when // an operation submitted with this object as its tag finishes. `result` is the // op result (>=0 byte count / fd, <0 -errno); `flags` is a CompletionFlags -// bitset. +// bitset. Each storage operation needs its own tag until completion, including +// time spent queued for backend capacity. Complete() may reuse the tag for the +// next operation or destroy it; backends must read all per-operation state +// before dispatching and must not access the tag afterwards. class IoCompletion { public: virtual ~IoCompletion() = default; @@ -43,6 +46,14 @@ class IoCompletion { protected: std::coroutine_handle<> awaiting_{}; + + private: + friend class SpdkStorageBackend; + + // Status-only storage completions need the expected successful byte count + // (zero for flush). Keep it with the caller-owned tag so submitted commands + // need no separately allocated backend callback context. + int success_result_ = 0; }; } // namespace bycorf diff --git a/include/bycorf/io/spdk_storage.h b/include/bycorf/io/spdk_storage.h index 4e42439..7a36b6e 100644 --- a/include/bycorf/io/spdk_storage.h +++ b/include/bycorf/io/spdk_storage.h @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -75,7 +76,9 @@ void FreeStorageBuffer(void* buffer, std::size_t alignment) noexcept; // Worker-owned asynchronous storage. Successful submissions retain access to // the caller's buffer/tag until completion, including while waiting for qpair -// capacity. Poll drives both device completions and deferred submissions. +// capacity. The backend owns waiting request metadata; each tag represents one +// unfinished request. Submit methods never dispatch completions inline. Poll +// drives both device completions and deferred submissions. class SpdkStorageBackend { public: SpdkStorageBackend() = default; @@ -111,12 +114,10 @@ class SpdkStorageBackend { void CompleteRequest(IoCompletion* tag, int result); private: - struct AsyncRequest { + struct PendingIo { IoCompletion* tag_ = nullptr; - AsyncRequest* next_ = nullptr; void* buffer_ = nullptr; std::uint64_t lba_ = 0; - int success_result_ = 0; std::uint32_t count_ = 0; // Only retries resolve this index. The fixed-file table cannot be resized // or an open entry closed while accepted I/O (including deferred I/O) @@ -132,38 +133,36 @@ class SpdkStorageBackend { struct OpenFile { void* device_ = nullptr; spdk_nvme_qpair* qpair_ = nullptr; + std::size_t channel_index_ = 0; }; struct ControllerChannel { void* controller_ = nullptr; spdk_nvme_qpair* qpair_ = nullptr; unsigned open_files_ = 0; - AsyncRequest* pending_head_ = nullptr; - AsyncRequest* pending_tail_ = nullptr; + std::deque pending_{}; }; absl::Status SubmitIo(FixedFile file, void* buffer, std::size_t bytes, std::uint64_t offset, bool write, IoCompletion* tag); - // Accepted requests retain their caller-owned buffers until completion, - // including time spent waiting for an SPDK request descriptor. - int TrySubmit(AsyncRequest* request, spdk_nvme_ns* ns, + int TrySubmit(const PendingIo& request, spdk_nvme_ns* ns, spdk_nvme_qpair* qpair); - absl::Status SubmitOrDefer(AsyncRequest* request, spdk_nvme_ns* ns, - spdk_nvme_qpair* qpair); - void RetryDeferred(ControllerChannel& channel, unsigned limit, + absl::Status SubmitOrDefer(const PendingIo& request, spdk_nvme_ns* ns, + std::size_t channel_index); + absl::Status QueuePending(const PendingIo& request, + std::size_t channel_index); + void RetryDeferred(std::size_t channel_index, unsigned limit, SpdkPollResult& result); OpenFile* Lookup(FixedFile file); - AsyncRequest* AcquireRequest() noexcept; - void ReleaseRequest(AsyncRequest* request) noexcept; static void CompleteAsync(void* context, const spdk_nvme_cpl* completion); Worker* worker_ = nullptr; std::vector files_; std::vector channels_; std::vector> pending_completions_; - std::vector request_pool_; - AsyncRequest* free_requests_ = nullptr; std::size_t next_poll_channel_ = 0; + // Completion callbacks may open/close files and invalidate channel indices. + std::size_t channel_generation_ = 0; std::size_t outstanding_ = 0; bool initialized_ = false; }; diff --git a/include/bycorf/runtime/worker.h b/include/bycorf/runtime/worker.h index 360193e..f6d8ca4 100644 --- a/include/bycorf/runtime/worker.h +++ b/include/bycorf/runtime/worker.h @@ -296,6 +296,9 @@ class Worker { #endif return backend_.RegisterBuffers(buffers); } + // Storage submissions retain the tag and referenced caller-owned buffers + // until completion, including while queued. Each tag may have only one + // unfinished storage operation; its completion may submit the next one. absl::Status SubmitOpenDirect(std::string_view path, int flags, mode_t mode, FixedFile file, IoCompletion* tag) { #if BYCORF_KERNEL_BYPASS diff --git a/src/io/spdk_storage.cpp b/src/io/spdk_storage.cpp index 4ef1060..d93e09c 100644 --- a/src/io/spdk_storage.cpp +++ b/src/io/spdk_storage.cpp @@ -36,6 +36,7 @@ #include #include +#include "absl/base/attributes.h" #include "bycorf/io/dpdk_environment.h" #include "bycorf/runtime/worker.h" @@ -44,6 +45,17 @@ namespace { constexpr std::string_view kSpdkPrefix = "spdk://"; +ABSL_ATTRIBUTE_COLD ABSL_ATTRIBUTE_NOINLINE absl::Status SubmitError(int rc) { + return absl::Status(absl::StatusCode::kUnavailable, + "SPDK I/O submission failed: rc=" + std::to_string(rc)); +} + +ABSL_ATTRIBUTE_COLD ABSL_ATTRIBUTE_NOINLINE absl::Status +PendingAllocationError() { + return absl::Status(absl::StatusCode::kResourceExhausted, + "SPDK pending request allocation failed"); +} + struct ParsedPath { std::string traddr_; std::uint32_t nsid_ = 1; @@ -373,20 +385,21 @@ void FreeStorageBuffer(void* buffer, std::size_t alignment) noexcept { void SpdkStorageBackend::CompleteAsync(void* context, const spdk_nvme_cpl* completion) { - auto* request = static_cast(context); + auto* tag = static_cast(context); const int result = - spdk_nvme_cpl_is_error(completion) ? -EIO : request->success_result_; + spdk_nvme_cpl_is_error(completion) ? -EIO : tag->success_result_; Worker* worker = ThisWorker().self_; assert(worker != nullptr); SpdkStorageBackend* backend = &worker->storage_backend_; - IoCompletion* tag = request->tag_; + // Complete may resubmit this tag or destroy it. Read its result first and + // never access caller-owned state after dispatching the terminal completion. backend->CompleteRequest(tag, result); - backend->ReleaseRequest(request); } SpdkStorageBackend::~SpdkStorageBackend() { Shutdown(); } void SpdkStorageBackend::CompleteRequest(IoCompletion* tag, int result) { + assert(outstanding_ != 0); --outstanding_; tag->Complete(*worker_, result, 0); } @@ -397,15 +410,6 @@ absl::Status SpdkStorageBackend::Init(Worker* worker) { "SPDK storage backend must be owned by its worker"); } worker_ = worker; - // Bound accepted foreground/background I/O, including requests waiting for - // qpair capacity, without a hot-path malloc. Read-buffer counts may exceed - // this bound; callers still receive ResourceExhausted if this pool is full. - request_pool_.resize(4096); - free_requests_ = nullptr; - for (AsyncRequest& request : request_pool_) { - request.next_ = free_requests_; - free_requests_ = &request; - } initialized_ = true; return absl::OkStatus(); } @@ -414,7 +418,7 @@ void SpdkStorageBackend::Shutdown() { if (!initialized_) { return; } - while (outstanding_ != 0) { + while (HasOutstanding()) { Poll(); } for (ControllerChannel& channel : channels_) { @@ -426,9 +430,8 @@ void SpdkStorageBackend::Shutdown() { files_.clear(); channels_.clear(); pending_completions_.clear(); - request_pool_.clear(); - free_requests_ = nullptr; next_poll_channel_ = 0; + channel_generation_ = 0; worker_ = nullptr; initialized_ = false; } @@ -464,23 +467,6 @@ SpdkStorageBackend::OpenFile* SpdkStorageBackend::Lookup(FixedFile file) { : nullptr; } -SpdkStorageBackend::AsyncRequest* -SpdkStorageBackend::AcquireRequest() noexcept { - AsyncRequest* request = free_requests_; - if (request != nullptr) { - free_requests_ = request->next_; - request->next_ = nullptr; - } - return request; -} - -void SpdkStorageBackend::ReleaseRequest(AsyncRequest* request) noexcept { - request->tag_ = nullptr; - request->success_result_ = 0; - request->next_ = free_requests_; - free_requests_ = request; -} - absl::Status SpdkStorageBackend::SubmitOpenDirect(std::string_view path, int flags, mode_t, FixedFile file, @@ -512,10 +498,12 @@ absl::Status SpdkStorageBackend::SubmitOpenDirect(std::string_view path, } channels_.push_back(ControllerChannel{ .controller_ = controller, .qpair_ = qpair, .open_files_ = 0}); + ++channel_generation_; channel = std::prev(channels_.end()); } opened.device_ = *device; opened.qpair_ = channel->qpair_; + opened.channel_index_ = static_cast(channel - channels_.begin()); ++channel->open_files_; // The coroutine awaitable stores its continuation only after this submit // method returns. Defer even immediately completed operations until Poll(). @@ -530,14 +518,20 @@ absl::Status SpdkStorageBackend::SubmitCloseDirect(FixedFile file, return absl::Status(absl::StatusCode::kFailedPrecondition, "invalid or busy SPDK close request"); } - auto channel = std::find_if(channels_.begin(), channels_.end(), - [opened](const ControllerChannel& candidate) { - return candidate.qpair_ == opened->qpair_; - }); - assert(channel != channels_.end() && channel->open_files_ != 0); - if (--channel->open_files_ == 0) { - spdk_nvme_ctrlr_free_io_qpair(channel->qpair_); - channels_.erase(channel); + const std::size_t index = opened->channel_index_; + ControllerChannel& channel = channels_[index]; + assert(channel.qpair_ == opened->qpair_ && channel.open_files_ != 0); + if (--channel.open_files_ == 0) { + spdk_nvme_ctrlr_free_io_qpair(channel.qpair_); + channels_.erase(channels_.begin() + index); + ++channel_generation_; + // Close is the only operation that shifts existing channel indices; + // ordinary submissions use the cached index without a controller search. + for (OpenFile& other : files_) { + if (other.qpair_ != nullptr && other.channel_index_ > index) { + --other.channel_index_; + } + } if (channels_.empty()) { next_poll_channel_ = 0; } else if (next_poll_channel_ >= channels_.size()) { @@ -560,28 +554,24 @@ absl::Status SpdkStorageBackend::SubmitIo(FixedFile file, void* buffer, } auto* device = static_cast(opened->device_); const std::uint32_t sector = device->sector_size_; - if (offset % sector != 0 || bytes % sector != 0 || + if (bytes > static_cast(std::numeric_limits::max()) || + offset % sector != 0 || bytes % sector != 0 || bytes / sector > std::numeric_limits::max() || offset > device->size_bytes_ || bytes > device->size_bytes_ - offset) { return absl::Status(absl::StatusCode::kInvalidArgument, - "unaligned SPDK I/O request"); - } - AsyncRequest* request = AcquireRequest(); - if (request == nullptr) { - return absl::Status(absl::StatusCode::kResourceExhausted, - "SPDK request pool exhausted"); - } - request->tag_ = tag; - request->success_result_ = static_cast(bytes); - request->file_ = file; - request->buffer_ = buffer; - request->lba_ = offset / sector; - request->count_ = static_cast(bytes / sector); - request->operation_ = - write ? AsyncRequest::Operation::kWrite : AsyncRequest::Operation::kRead; + "invalid SPDK I/O range or alignment"); + } + const PendingIo request{.tag_ = tag, + .buffer_ = buffer, + .lba_ = offset / sector, + .count_ = static_cast(bytes / sector), + .file_ = file, + .operation_ = write ? PendingIo::Operation::kWrite + : PendingIo::Operation::kRead}; + tag->success_result_ = static_cast(bytes); // Reuse the lookup needed for validation on the ordinary submission path; // only a deferred retry needs to resolve the saved file index again. - return SubmitOrDefer(request, device->ns_, opened->qpair_); + return SubmitOrDefer(request, device->ns_, opened->channel_index_); } absl::Status SpdkStorageBackend::SubmitReadFixed(FixedFile file, @@ -621,6 +611,7 @@ absl::Status SpdkStorageBackend::SubmitFdatasync(FixedFile file, "invalid SPDK flush request"); } auto* device = static_cast(opened->device_); + tag->success_result_ = 0; const auto* data = spdk_nvme_ctrlr_get_data(device->controller_->ctrlr_); if (!data->vwc.present) { // NVMe defines FLUSH as a no-op without a volatile write cache. Its scope @@ -632,91 +623,93 @@ absl::Status SpdkStorageBackend::SubmitFdatasync(FixedFile file, pending_completions_.emplace_back(tag, 0); return absl::OkStatus(); } - AsyncRequest* request = AcquireRequest(); - if (request == nullptr) { - return absl::Status(absl::StatusCode::kResourceExhausted, - "SPDK request pool exhausted"); - } - request->tag_ = tag; - request->success_result_ = 0; - request->file_ = file; - request->operation_ = AsyncRequest::Operation::kFlush; - return SubmitOrDefer(request, device->ns_, opened->qpair_); + const PendingIo request{ + .tag_ = tag, .file_ = file, .operation_ = PendingIo::Operation::kFlush}; + return SubmitOrDefer(request, device->ns_, opened->channel_index_); } -int SpdkStorageBackend::TrySubmit(AsyncRequest* request, spdk_nvme_ns* ns, +int SpdkStorageBackend::TrySubmit(const PendingIo& request, spdk_nvme_ns* ns, spdk_nvme_qpair* qpair) { - switch (request->operation_) { - case AsyncRequest::Operation::kRead: - return spdk_nvme_ns_cmd_read(ns, qpair, request->buffer_, request->lba_, - request->count_, CompleteAsync, request, 0); - case AsyncRequest::Operation::kWrite: - return spdk_nvme_ns_cmd_write(ns, qpair, request->buffer_, request->lba_, - request->count_, CompleteAsync, request, 0); - case AsyncRequest::Operation::kFlush: - return spdk_nvme_ns_cmd_flush(ns, qpair, CompleteAsync, request); + switch (request.operation_) { + case PendingIo::Operation::kRead: + return spdk_nvme_ns_cmd_read(ns, qpair, request.buffer_, request.lba_, + request.count_, CompleteAsync, request.tag_, + 0); + case PendingIo::Operation::kWrite: + return spdk_nvme_ns_cmd_write(ns, qpair, request.buffer_, request.lba_, + request.count_, CompleteAsync, request.tag_, + 0); + case PendingIo::Operation::kFlush: + return spdk_nvme_ns_cmd_flush(ns, qpair, CompleteAsync, request.tag_); } return -EINVAL; } -absl::Status SpdkStorageBackend::SubmitOrDefer(AsyncRequest* request, +absl::Status SpdkStorageBackend::SubmitOrDefer(const PendingIo& request, spdk_nvme_ns* ns, - spdk_nvme_qpair* qpair) { + std::size_t channel_index) { + const ControllerChannel& channel = channels_[channel_index]; + if (!channel.pending_.empty()) [[unlikely]] { + // Preserve submission order even if device capacity has become available + // since an earlier request joined this qpair's software queue. + return QueuePending(request, channel_index); + } + spdk_nvme_qpair* qpair = channel.qpair_; const int rc = TrySubmit(request, ns, qpair); if (rc == 0) [[likely]] { ++outstanding_; return absl::OkStatus(); } if (rc == -ENOMEM && spdk_nvme_qpair_get_num_outstanding_reqs(qpair) != 0) { - // SPDK request descriptors are bounded independently of our DMA slots. - // A rejected command has not taken ownership of the buffer or callback. - // Keep both alive in the worker-local pool and retry after polling, never - // poll inline: the awaitable has not installed its continuation yet. - auto channel = std::find_if( - channels_.begin(), channels_.end(), - [qpair](const auto& candidate) { return candidate.qpair_ == qpair; }); - assert(channel != channels_.end()); - if (channel->pending_tail_ != nullptr) { - channel->pending_tail_->next_ = request; - } else { - channel->pending_head_ = request; - } - channel->pending_tail_ = request; - // Count deferred commands too: close/shutdown and worker sleep must not - // discard a buffer or qpair while a command is waiting to be submitted. - ++outstanding_; - return absl::OkStatus(); + return QueuePending(request, channel_index); } - ReleaseRequest(request); - return absl::Status(absl::StatusCode::kUnavailable, - "SPDK I/O submission failed: rc=" + std::to_string(rc)); + return SubmitError(rc); } -void SpdkStorageBackend::RetryDeferred(ControllerChannel& channel, - unsigned limit, SpdkPollResult& result) { - for (unsigned submitted = 0; - channel.pending_head_ != nullptr && (limit == 0 || submitted < limit); +ABSL_ATTRIBUTE_NOINLINE absl::Status SpdkStorageBackend::QueuePending( + const PendingIo& request, std::size_t channel_index) { + // SPDK retains nothing when submission fails. Store only unsubmitted + // arguments here; issued requests use the caller's tag as callback context. + // Never poll inline: await_suspend has not installed its continuation yet. + try { + channels_[channel_index].pending_.push_back(request); + } catch (const std::bad_alloc&) { + return PendingAllocationError(); + } + // Admission, rather than device submission, owns this count. Close, sleep + // and shutdown must also account for requests waiting for device capacity. + ++outstanding_; + return absl::OkStatus(); +} + +ABSL_ATTRIBUTE_NOINLINE void SpdkStorageBackend::RetryDeferred( + std::size_t channel_index, unsigned limit, SpdkPollResult& result) { + const std::size_t generation = channel_generation_; + for (unsigned submitted = 0; !channels_[channel_index].pending_.empty() && + (limit == 0 || submitted < limit); ++submitted) { - AsyncRequest* request = channel.pending_head_; - const OpenFile& opened = files_[request->file_.index_]; - assert(opened.qpair_ == channel.qpair_); + // Callback dispatch below can change channels_ or enqueue another request. + // Keep values across it, never references into either container. + const PendingIo request = channels_[channel_index].pending_.front(); + const OpenFile opened = files_[request.file_.index_]; + spdk_nvme_qpair* qpair = channels_[channel_index].qpair_; + assert(opened.qpair_ == qpair); auto* device = static_cast(opened.device_); - const int rc = TrySubmit(request, device->ns_, channel.qpair_); - if (rc == -ENOMEM && - spdk_nvme_qpair_get_num_outstanding_reqs(channel.qpair_) != 0) { + const int rc = TrySubmit(request, device->ns_, qpair); + if (rc == -ENOMEM && spdk_nvme_qpair_get_num_outstanding_reqs(qpair) != 0) { break; } - channel.pending_head_ = request->next_; - if (channel.pending_head_ == nullptr) channel.pending_tail_ = nullptr; - request->next_ = nullptr; + // A successful command retains only request.tag_, so its queue storage can + // be reclaimed immediately. Errors also leave the FIFO before dispatch. + channels_[channel_index].pending_.pop_front(); result.did_work_ = true; if (rc != 0) { // With no in-flight command on this qpair, ENOMEM cannot be cured by // waiting for a completion (e.g. an unsplittable oversized request). // Surface that failure instead of stranding shutdown forever. - CompleteRequest(request->tag_, rc); - ReleaseRequest(request); ++result.completions_; + CompleteRequest(request.tag_, rc); + if (channel_generation_ != generation) return; } } } @@ -734,13 +727,13 @@ SpdkPollResult SpdkStorageBackend::Poll(unsigned max_completions) { result.completions_ = static_cast(completions.size()); } const std::size_t channel_count = channels_.size(); + const std::size_t generation = channel_generation_; // Keep this pass's origin fixed while advancing the next-pass cursor below. // Using the mutable cursor in the index skips channels (with six channels, // two qpairs can remain unpolled forever and stall reads or flushes). const std::size_t first_channel = next_poll_channel_; for (std::size_t visited = 0; visited < channel_count; ++visited) { const std::size_t index = (first_channel + visited) % channel_count; - ControllerChannel& channel = channels_[index]; const unsigned remaining = max_completions == 0 ? 0 : (result.completions_ >= max_completions @@ -750,16 +743,22 @@ SpdkPollResult SpdkStorageBackend::Poll(unsigned max_completions) { next_poll_channel_ = index; break; } + spdk_nvme_qpair* qpair = channels_[index].qpair_; + next_poll_channel_ = (index + 1) % channel_count; const int32_t completed = - spdk_nvme_qpair_process_completions(channel.qpair_, remaining); + spdk_nvme_qpair_process_completions(qpair, remaining); if (completed > 0) { result.did_work_ = true; result.completions_ += static_cast(completed); } - if (channel.pending_head_ != nullptr) [[unlikely]] { - RetryDeferred(channel, max_completions, result); + // Caller callbacks can open or close files, moving channels_ or deleting + // this qpair. SPDK defers a callback's qpair free until its poll returns; + // resume our traversal on the next pass using the updated channel table. + if (channel_generation_ != generation) break; + if (!channels_[index].pending_.empty()) [[unlikely]] { + RetryDeferred(index, max_completions, result); + if (channel_generation_ != generation) break; } - next_poll_channel_ = (index + 1) % channel_count; } return result; } diff --git a/tests/spdk_backpressure_stub_check.cpp b/tests/spdk_backpressure_stub_check.cpp index cab3acc..a2bbb2f 100644 --- a/tests/spdk_backpressure_stub_check.cpp +++ b/tests/spdk_backpressure_stub_check.cpp @@ -16,7 +16,7 @@ // Exercise the production backend and Worker with link-time SPDK stubs. Only // device discovery, environment initialization and NVMe commands are replaced; -// the request pool, pending FIFO, polling and callback ownership stay real. +// the pending FIFO, polling and callback ownership stay real. #include #include #include @@ -29,6 +29,7 @@ #include #include #include +#include #include #include #include @@ -85,6 +86,7 @@ struct Qpair { }; std::vector qpairs; unsigned environment_calls = 0; +bool volatile_write_cache = true; Qpair& Queue(spdk_nvme_qpair* qpair) { auto* queue = reinterpret_cast(qpair); @@ -135,14 +137,15 @@ spdk_nvme_ns* __wrap_spdk_nvme_ctrlr_get_ns(spdk_nvme_ctrlr* ctrlr, } bool __wrap_spdk_nvme_ns_is_active(spdk_nvme_ns*) { return true; } uint32_t __wrap_spdk_nvme_ns_get_sector_size(spdk_nvme_ns*) { return 512; } -uint64_t __wrap_spdk_nvme_ns_get_size(spdk_nvme_ns*) { return 1ULL << 30; } +uint64_t __wrap_spdk_nvme_ns_get_size(spdk_nvme_ns*) { return 1ULL << 34; } const spdk_nvme_ctrlr_data* __wrap_spdk_nvme_ctrlr_get_data(spdk_nvme_ctrlr*) { static const auto data = [] { spdk_nvme_ctrlr_data value{}; value.vwc.present = 1; return value; }(); - return &data; + static const spdk_nvme_ctrlr_data no_cache{}; + return volatile_write_cache ? &data : &no_cache; } spdk_nvme_qpair* __wrap_spdk_nvme_ctrlr_alloc_io_qpair( spdk_nvme_ctrlr* ctrlr, const spdk_nvme_io_qpair_opts*, size_t) { @@ -251,6 +254,8 @@ void CheckQueues(bycorf::Worker& worker, std::span buffer) { other_deferred.calls == 0); CHECK(first.accepted.size() == 1 && second.accepted.size() == 1); + CHECK(first.accepted[0].context == &read); + CHECK(second.accepted[0].context == &other); unsigned attempts = first.attempts; for (unsigned i = 0; i < 3; ++i) worker.RunOnce(false); CHECK(first.attempts > attempts); @@ -267,6 +272,7 @@ void CheckQueues(bycorf::Worker& worker, std::span buffer) { CHECK(read.calls + write.calls + flush.calls == 0); CHECK(first.accepted.size() == 1 && second.accepted.size() == 2); CHECK(second.accepted[1].operation == Operation::kRead); + CHECK(second.accepted[1].context == &other_deferred); CHECK(second.accepted[1].ns == reinterpret_cast(&controllers[1].namespaces[0])); CHECK(second.accepted[1].buffer == buffer.data() && @@ -278,6 +284,7 @@ void CheckQueues(bycorf::Worker& worker, std::span buffer) { CHECK(write.calls == 0 && flush.calls == 0); const auto& retried = first.accepted[1]; CHECK(retried.operation == Operation::kWrite); + CHECK(retried.context == &write); CHECK(retried.ns == reinterpret_cast(&controllers[0].namespaces[1])); CHECK(retried.buffer == buffer.data() && retried.lba == 2 && @@ -287,6 +294,7 @@ void CheckQueues(bycorf::Worker& worker, std::span buffer) { CHECK(write.calls == 1 && write.result == static_cast(buffer.size())); CHECK(flush.calls == 0); CHECK(first.accepted[2].operation == Operation::kFlush); + CHECK(first.accepted[2].context == &flush); CHECK(first.accepted[2].ns == reinterpret_cast(&controllers[0].namespaces[0])); first.completion_tokens = 1; @@ -298,12 +306,26 @@ void CheckQueues(bycorf::Worker& worker, std::span buffer) { void CheckErrors(bycorf::Worker& worker, std::span buffer) { auto& queue = *qpairs[0]; + Completion oversized; + const auto attempts = queue.attempts; + // The stub namespace is larger than this sector-aligned request, but its + // successful byte count cannot fit in IoCompletion's int result. Stub I/O + // never touches the buffer; validation must reject it before SPDK is called. + CHECK(worker + .SubmitReadFixed({0}, + {.data_ = buffer.data(), + .size_ = static_cast( + std::numeric_limits::max()) + + 1}, + 0, &oversized) + .code() == absl::StatusCode::kInvalidArgument); + CHECK(oversized.calls == 0 && queue.attempts == attempts); for (int error : {-EINVAL, -ENOMEM}) { queue.submit_error = error; Completion rejected; - // More than the pool capacity catches a request leaked on synchronous - // rejection. An empty qpair must not defer even an ENOMEM failure. - for (unsigned i = 0; i < 4097; ++i) { + // An empty qpair cannot make progress through a completion, even for + // ENOMEM. Rejection must not retain the tag or poison later submissions. + for (unsigned i = 0; i < 3; ++i) { CHECK(worker.SubmitRead({0}, buffer, 0, &rejected).code() == absl::StatusCode::kUnavailable); } @@ -338,24 +360,252 @@ void CheckErrors(bycorf::Worker& worker, std::span buffer) { "completion errors\n"; } -void CheckPool(bycorf::Worker& worker, std::span buffer) { - for (unsigned round = 0; round < 2; ++round) { - std::vector tags(4096); - for (auto& tag : tags) CHECK(worker.SubmitRead({0}, buffer, 0, &tag).ok()); - for (auto& tag : tags) CHECK(tag.calls == 0); - Completion rejected, closed; - CHECK(worker.SubmitRead({0}, buffer, 0, &rejected).code() == - absl::StatusCode::kResourceExhausted); +void CheckLargeBursts(bycorf::Worker& worker, std::span buffer) { + auto& first = *qpairs[0]; + auto& second = *qpairs[1]; + constexpr unsigned kCount = 8193; + // Exercise both a large software backlog and many hardware-accepted I/Os. + // Neither may impose the former worker-wide 4096-request limit on a peer. + for (unsigned capacity : {1U, kCount}) { + first.capacity = capacity; + const auto first_begin = first.accepted.size(); + const auto second_begin = second.accepted.size(); + std::vector tags(kCount); + for (std::size_t i = 0; i < tags.size(); ++i) { + const auto bytes = (i % 8 + 1) * 512; + CHECK( + worker.SubmitRead({0}, buffer.first(bytes), i * 512, &tags[i]).ok()); + } + for (const auto& tag : tags) CHECK(tag.calls == 0); + CHECK(first.accepted.size() == first_begin + capacity); + Completion other, closed; + CHECK(worker.SubmitRead({2}, buffer.first(512), 0, &other).ok()); + CHECK(other.calls == 0); + CHECK(second.accepted.size() == second_begin + 1); + CHECK(second.accepted.back().context == &other); CHECK(!worker.SubmitCloseDirect({0}, &closed).ok()); CHECK(!worker.RegisterFixedFiles(4).ok()); - qpairs[0]->completion_tokens = tags.size(); + + second.completion_tokens = 1; + Drain(worker, [&] { return other.calls == 1; }); + CHECK(other.result == 512); + for (const auto& tag : tags) CHECK(tag.calls == 0); + first.completion_tokens = tags.size(); Drain(worker, [&] { return tags.back().calls == 1; }); - for (auto& tag : tags) - CHECK(tag.calls == 1 && tag.result == static_cast(buffer.size())); - CHECK(rejected.calls == 0 && closed.calls == 0); - CHECK(qpairs[0]->inflight.empty()); + CHECK(first.accepted.size() == first_begin + tags.size()); + for (std::size_t i = 0; i < tags.size(); ++i) { + CHECK(tags[i].calls == 1 && + tags[i].result == static_cast((i % 8 + 1) * 512)); + const auto& command = first.accepted[first_begin + i]; + CHECK(command.context == &tags[i]); + CHECK(command.lba == i && command.count == i % 8 + 1); + } + CHECK(closed.calls == 0); + CHECK(first.inflight.empty() && second.inflight.empty()); + CHECK(first.completion_tokens == 0 && second.completion_tokens == 0); + } + first.capacity = 1; + std::cout + << "PASS large queued/issued bursts and independent qpair progress\n"; +} + +void CheckLateSubmissionFifo(bycorf::Worker& worker, + std::span buffer) { + auto& queue = *qpairs[0]; + const auto begin = queue.accepted.size(); + std::array tags; + for (unsigned i = 0; i < 3; ++i) + CHECK(worker.SubmitRead({0}, buffer, i * 512, &tags[i]).ok()); + CHECK(queue.accepted.size() == begin + 1); + queue.capacity = 4; + const auto attempts = queue.attempts; + CHECK(worker.SubmitRead({0}, buffer, 1536, &tags.back()).ok()); + CHECK(queue.attempts == attempts); + for (const auto& tag : tags) CHECK(tag.calls == 0); + // Capacity is available again, but the late producer cannot skip the two + // older requests already accepted into this qpair's software FIFO. + queue.completion_tokens = 1; + Drain(worker, [&] { return queue.accepted.size() == begin + tags.size(); }); + CHECK(tags.front().calls == 1); + for (std::size_t i = 0; i < tags.size(); ++i) { + CHECK(queue.accepted[begin + i].lba == i); + CHECK(queue.accepted[begin + i].context == &tags[i]); + if (i != 0) CHECK(tags[i].calls == 0); } - std::cout << "PASS bounded pool, close protection and complete pool reuse\n"; + queue.completion_tokens = tags.size() - 1; + Drain(worker, [&] { return tags.back().calls == 1; }); + queue.capacity = 1; + std::cout << "PASS late submissions preserve FIFO when capacity returns\n"; +} + +void CheckCompletionOwnership(bycorf::Worker& worker, + std::span buffer) { + struct ReusedCompletion final : bycorf::IoCompletion { + explicit ReusedCompletion(std::span bytes) : buffer(bytes) {} + void Complete(bycorf::Worker& owner, int value, unsigned flags) override { + CHECK(&owner == bycorf::ThisWorker().self_ && flags == 0); + CHECK(calls < results.size()); + results[calls++] = value; + if (calls == 1) + CHECK(owner.SubmitWrite({1}, buffer.first(1024), 512, this).ok()); + else if (calls == 2) + CHECK(owner.SubmitFdatasync({0}, this).ok()); + } + std::span buffer; + std::array results{}; + unsigned calls = 0; + } reused(buffer); + auto& queue = *qpairs[0]; + CHECK(worker.SubmitRead({0}, buffer.first(512), 0, &reused).ok()); + CHECK(reused.calls == 0 && queue.accepted.back().context == &reused); + for (unsigned expected_calls = 1; expected_calls <= 3; ++expected_calls) { + queue.completion_tokens = 1; + Drain(worker, [&] { return reused.calls == expected_calls; }); + CHECK(queue.accepted.back().context == &reused); + } + CHECK((reused.results == std::array{512, 1024, 0})); + + struct SelfDeletingCompletion final : bycorf::IoCompletion { + explicit SelfDeletingCompletion(bool& done) : destroyed(done) {} + ~SelfDeletingCompletion() override { destroyed = true; } + void Complete(bycorf::Worker& owner, int result, unsigned flags) override { + CHECK(&owner == bycorf::ThisWorker().self_); + CHECK(result == 512 && flags == 0); + delete this; + } + bool& destroyed; + }; + bool destroyed = false; + auto* self_deleting = new SelfDeletingCompletion(destroyed); + CHECK(worker.SubmitRead({0}, buffer.first(512), 0, self_deleting).ok()); + CHECK(!destroyed && queue.accepted.back().context == self_deleting); + queue.completion_tokens = 1; + Drain(worker, [&] { return destroyed; }); + std::cout + << "PASS callback tag reuse, byte/flush results and self-destruction\n"; +} + +void CheckRetryErrorReentry(bycorf::Worker& worker, + std::span buffer) { + struct ReusedCompletion final : bycorf::IoCompletion { + explicit ReusedCompletion(std::span bytes) : buffer(bytes) {} + void Complete(bycorf::Worker& owner, int result, unsigned flags) override { + CHECK(&owner == bycorf::ThisWorker().self_ && flags == 0); + CHECK(++calls <= 2); + if (calls == 1) { + CHECK(result == -EINVAL); + qpairs[0]->submit_error = 0; + CHECK(owner.SubmitRead({0}, buffer.first(1024), 1024, this).ok()); + } else { + CHECK(result == 1024); + } + } + std::span buffer; + unsigned calls = 0; + } reused(buffer); + auto& queue = *qpairs[0]; + const auto begin = queue.accepted.size(); + Completion active, older; + CHECK(worker.SubmitRead({0}, buffer, 0, &active).ok()); + CHECK(worker.SubmitRead({0}, buffer.first(2048), 512, &reused).ok()); + CHECK(worker.SubmitRead({0}, buffer, 1536, &older).ok()); + CHECK(active.calls + reused.calls + older.calls == 0); + queue.submit_error = -EINVAL; + Drain(worker, [&] { return reused.calls == 1; }); + CHECK(active.calls == 0 && older.calls == 0); + CHECK(queue.accepted.size() == begin + 1); + queue.completion_tokens = 1; + Drain(worker, [&] { return queue.accepted.size() == begin + 2; }); + CHECK(active.calls == 1 && older.calls == 0 && reused.calls == 1); + CHECK(queue.accepted.back().context == &older); + queue.completion_tokens = 1; + Drain(worker, [&] { return queue.accepted.size() == begin + 3; }); + CHECK(older.calls == 1 && reused.calls == 1); + CHECK(queue.accepted.back().context == &reused); + CHECK(queue.accepted.back().count == 2 && queue.accepted.back().lba == 2); + queue.completion_tokens = 1; + Drain(worker, [&] { return reused.calls == 2; }); + std::cout << "PASS retry-error callback reentry preserves older queued I/O\n"; +} + +void CheckChannelReopen(bycorf::Worker& worker, std::span buffer) { + std::array closed; + CHECK(worker.SubmitCloseDirect({0}, &closed[0]).ok()); + CHECK(worker.SubmitCloseDirect({1}, &closed[1]).ok()); + CHECK(closed[0].calls + closed[1].calls == 0); + CHECK(qpairs.size() == 1); + Drain(worker, [&] { return closed[1].calls == 1; }); + CHECK(closed[0].calls == 1); + + struct OpeningCompletion final : Completion { + void Complete(bycorf::Worker& owner, int result, unsigned flags) override { + Completion::Complete(owner, result, flags); + for (unsigned i = 0; i < opened.size(); ++i) + CHECK( + owner.SubmitOpenDirect(paths[i], O_RDWR, 0, {i}, &opened[i]).ok()); + } + std::array opened; + } opening; + // Closing the first controller shifts the remaining file's channel index. + // Opening a controller from its completion then changes the channel table + // during polling, while the qpair being polled remains valid. + CHECK(worker.SubmitRead({2}, buffer.first(512), 0, &opening).ok()); + CHECK(qpairs[0]->accepted.back().context == &opening); + qpairs[0]->completion_tokens = 1; + Drain(worker, [&] { return opening.opened.back().calls == 1; }); + CHECK(opening.calls == 1 && opening.result == 512); + CHECK(opening.opened[0].calls == 1 && qpairs.size() == 2); + + Completion remaining, reopened; + CHECK(worker.SubmitRead({2}, buffer.first(512), 0, &remaining).ok()); + CHECK(worker.SubmitRead({0}, buffer.first(1024), 0, &reopened).ok()); + CHECK(qpairs[0]->accepted.back().context == &remaining); + CHECK(qpairs[1]->accepted.back().context == &reopened); + qpairs[0]->completion_tokens = 1; + qpairs[1]->completion_tokens = 1; + Drain(worker, [&] { return remaining.calls + reopened.calls == 2; }); + CHECK(remaining.result == 512 && reopened.result == 1024); + + Completion closed_other, shifted, opened_other; + CHECK(worker.SubmitCloseDirect({2}, &closed_other).ok()); + CHECK(worker.SubmitRead({1}, buffer.first(2048), 0, &shifted).ok()); + CHECK(qpairs.size() == 1 && qpairs[0]->accepted.back().context == &shifted); + qpairs[0]->completion_tokens = 1; + Drain(worker, [&] { return shifted.calls == 1; }); + CHECK(closed_other.calls == 1 && shifted.result == 2048); + CHECK(worker.SubmitOpenDirect(paths[2], O_RDWR, 0, {2}, &opened_other).ok()); + Drain(worker, [&] { return opened_other.calls == 1; }); + CHECK(qpairs.size() == 2); + std::cout << "PASS channel removal, callback open and shifted file indices\n"; +} + +void CheckSyntheticShutdown(bycorf::Worker& worker) { + for (unsigned operation = 0; operation < 3; ++operation) { + CHECK(worker.Init({.recv_buffer_count_ = 0}).ok()); + CHECK(worker.RegisterFixedFiles(1).ok()); + Completion opened, completed; + CHECK(worker.SubmitOpenDirect(paths[0], O_RDWR, 0, {0}, &opened).ok()); + CHECK(opened.calls == 0); + if (operation != 0) { + Drain(worker, [&] { return opened.calls == 1; }); + if (operation == 1) { + CHECK(worker.SubmitCloseDirect({0}, &completed).ok()); + } else { + volatile_write_cache = false; + CHECK(worker.SubmitFdatasync({0}, &completed).ok()); + } + CHECK(completed.calls == 0); + } + // No NVMe command is in flight. Shutdown still owes the accepted open, + // close or no-VWC flush its deferred completion on the owning worker. + for (const auto* queue : qpairs) CHECK(queue->inflight.empty()); + worker.Shutdown(); + CHECK(opened.calls == 1 && opened.result == 0); + if (operation != 0) CHECK(completed.calls == 1 && completed.result == 0); + CHECK(qpairs.empty()); + volatile_write_cache = true; + } + std::cout << "PASS shutdown delivers synthetic open/close/no-VWC flush\n"; } } // namespace @@ -387,18 +637,30 @@ int main() { std::array buffer{}; CheckQueues(worker, buffer); CheckErrors(worker, buffer); - CheckPool(worker, buffer); + CheckLargeBursts(worker, buffer); + CheckLateSubmissionFifo(worker, buffer); + CheckCompletionOwnership(worker, buffer); + CheckRetryErrorReentry(worker, buffer); + CheckChannelReopen(worker, buffer); - Completion read, deferred; - CHECK(worker.SubmitRead({0}, buffer, 0, &read).ok()); - CHECK(worker.SubmitWrite({0}, buffer, 0, &deferred).ok()); - CHECK(read.calls + deferred.calls == 0); - qpairs[0]->completion_tokens = 2; + std::vector shutdown(5000); + for (std::size_t i = 0; i < shutdown.size(); ++i) { + CHECK((i % 2 == 0 ? worker.SubmitRead({0}, buffer, 0, &shutdown[i]) + : worker.SubmitWrite({0}, buffer, 0, &shutdown[i])) + .ok()); + CHECK(shutdown[i].calls == 0); + } + Completion other; + CHECK(worker.SubmitRead({2}, std::span(buffer).first(512), 0, &other).ok()); + CHECK(other.calls == 0); + qpairs[0]->completion_tokens = shutdown.size(); + qpairs[1]->completion_tokens = 1; worker.Shutdown(); - CHECK(read.calls == 1 && read.result == 4096); - CHECK(deferred.calls == 1 && deferred.result == 4096); + for (const auto& tag : shutdown) CHECK(tag.calls == 1 && tag.result == 4096); + CHECK(other.calls == 1 && other.result == 512); CHECK(qpairs.empty()); - close(wake_fd); std::cout << "PASS shutdown drains in-flight and deferred I/O on the owning " "worker\n"; + CheckSyntheticShutdown(worker); + close(wake_fd); } From b47fb60667a4c12e8ffacdbc661a681cc0530890 Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Fri, 16 Oct 2026 05:01:47 +0000 Subject: [PATCH 3/4] Drive deferred peer completions in io_uring wake test The MSG_RING target uses DEFER_TASKRUN. Polling CQEs alone can leave its task work pending on newer kernels; call Submit before Poll to match the worker loop. Keep the existing deadline and pre-submission assertions. --- tests/io_uring_backpressure_check.cpp | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/io_uring_backpressure_check.cpp b/tests/io_uring_backpressure_check.cpp index 72b8cb8..c6a1f77 100644 --- a/tests/io_uring_backpressure_check.cpp +++ b/tests/io_uring_backpressure_check.cpp @@ -104,6 +104,11 @@ void CheckSendsAndWake() { bool peer_woken = false; Drain(source.backend, [&] { + // DEFER_TASKRUN can defer MSG_RING delivery until the target ring enters + // the kernel, even when both rings share a thread. Drive the peer as its + // worker would: Submit runs deferred task work; Poll consumes ready CQEs. + const auto status = peer.backend.Submit(); + CHECK(status.ok() || absl::IsUnavailable(status)); peer_woken |= peer.backend.Poll(); for (const auto& tag : tags) if (tag.calls == 0) return false; From fd4c86eb95a00be241dbeba9e928a6dc22de079f Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Fri, 16 Oct 2026 07:42:25 +0000 Subject: [PATCH 4/4] Prepare io_uring requests directly in available SQEs Pass synchronous preparation callbacks to SubmitOrQueue so ordinary requests write into the ring instead of zeroing and copying a temporary 64-byte SQE. Keep queued descriptor preparation in a separate non-inlined frame, preserve FIFO admission, and share the returned Status with that slow path to avoid a protected temporary in GCC's common path. Track cancels at their call site. Keep timeout arguments caller-owned and allocate fallback receive buffers before advancing the SQ tail. Extend the small-ring regression with mixed send/sendmsg submissions across direct and queued admission. --- include/bycorf/io/io_uring_backend.h | 6 +- src/io/io_uring_backend.cpp | 329 +++++++++++++------------- tests/io_uring_backpressure_check.cpp | 17 +- 3 files changed, 182 insertions(+), 170 deletions(-) diff --git a/include/bycorf/io/io_uring_backend.h b/include/bycorf/io/io_uring_backend.h index be6df1d..4743181 100644 --- a/include/bycorf/io/io_uring_backend.h +++ b/include/bycorf/io/io_uring_backend.h @@ -176,8 +176,10 @@ class IoUringBackend { }; io_uring_sqe* AcquireSqe(); - absl::Status SubmitOrQueue(const io_uring_sqe& prepared); - absl::Status QueuePrepared(const io_uring_sqe& prepared); + template + absl::Status SubmitOrQueue(Prepare prepare); + template + void QueuePrepared(Prepare prepare, absl::Status* status); absl::Status SubmitSlow(); bool PollDeferred(); bool DrainPending(); diff --git a/src/io/io_uring_backend.cpp b/src/io/io_uring_backend.cpp index fbbebb9..4479947 100644 --- a/src/io/io_uring_backend.cpp +++ b/src/io/io_uring_backend.cpp @@ -245,6 +245,40 @@ void IoUringBackend::UnregisterStorageResources() { } } +template +absl::Status IoUringBackend::SubmitOrQueue(Prepare prepare) { + // Share one return object with the slow path. Returning its Status by value + // would otherwise introduce a protected stack temporary in the common path + // with GCC, even though the SQE itself is no longer stack-allocated here. + absl::Status status; + // Older accepted work retains priority. Preparation runs synchronously and + // is never retained; only the prepared descriptor can enter the queue. + // Never submit/reap here: callers install their continuation afterwards. + if (initialized_ && pending_io_.empty()) [[likely]] { + if (auto* sqe = AcquireSqe()) [[likely]] { + prepare(sqe); + return status; + } + } + QueuePrepared(prepare, &status); + return status; +} + +// Keep the temporary SQE and queue allocation in a separate frame, so ordinary +// submissions do not construct or copy a temporary SQE. Backlogs can be +// sustained, so this helper is not cold. +template +ABSL_ATTRIBUTE_NOINLINE void IoUringBackend::QueuePrepared( + Prepare prepare, absl::Status* status) { + if (!initialized_) { + *status = absl::FailedPreconditionError("backend is not initialized"); + return; + } + io_uring_sqe prepared{}; + prepare(&prepared); + pending_io_.push_back(PendingIo{prepared}); +} + absl::Status IoUringBackend::SubmitOpenDirect(std::string_view path, int flags, mode_t mode, FixedFile file, IoCompletion* tag) { @@ -252,94 +286,89 @@ absl::Status IoUringBackend::SubmitOpenDirect(std::string_view path, int flags, return absl::Status(absl::StatusCode::kFailedPrecondition, "fixed-file table is not registered"); } - io_uring_sqe prepared{}; - auto* sqe = &prepared; - io_uring_prep_openat_direct(sqe, AT_FDCWD, path.data(), flags, mode, - file.index_); - io_uring_sqe_set_data(sqe, tag); - return SubmitOrQueue(prepared); + return SubmitOrQueue([=](io_uring_sqe* sqe) { + io_uring_prep_openat_direct(sqe, AT_FDCWD, path.data(), flags, mode, + file.index_); + io_uring_sqe_set_data(sqe, tag); + }); } absl::Status IoUringBackend::SubmitCloseDirect(FixedFile file, IoCompletion* tag) { - io_uring_sqe prepared{}; - auto* sqe = &prepared; - io_uring_prep_close_direct(sqe, file.index_); - io_uring_sqe_set_data(sqe, tag); - return SubmitOrQueue(prepared); + return SubmitOrQueue([=](io_uring_sqe* sqe) { + io_uring_prep_close_direct(sqe, file.index_); + io_uring_sqe_set_data(sqe, tag); + }); } absl::Status IoUringBackend::SubmitReadFixed(FixedFile file, FixedBuffer buffer, std::uint64_t offset, IoCompletion* tag) { - io_uring_sqe prepared{}; - auto* sqe = &prepared; - io_uring_prep_read_fixed(sqe, static_cast(file.index_), buffer.data_, - static_cast(buffer.size_), offset, - buffer.index_); - sqe->flags |= IOSQE_FIXED_FILE; - io_uring_sqe_set_data(sqe, tag); - return SubmitOrQueue(prepared); + return SubmitOrQueue([=](io_uring_sqe* sqe) { + io_uring_prep_read_fixed(sqe, static_cast(file.index_), buffer.data_, + static_cast(buffer.size_), offset, + buffer.index_); + sqe->flags |= IOSQE_FIXED_FILE; + io_uring_sqe_set_data(sqe, tag); + }); } absl::Status IoUringBackend::SubmitRead(FixedFile file, std::span buffer, std::uint64_t offset, IoCompletion* tag) { - io_uring_sqe prepared{}; - auto* sqe = &prepared; - io_uring_prep_read(sqe, static_cast(file.index_), buffer.data(), - static_cast(buffer.size()), offset); - sqe->flags |= IOSQE_FIXED_FILE; - io_uring_sqe_set_data(sqe, tag); - return SubmitOrQueue(prepared); + return SubmitOrQueue([=](io_uring_sqe* sqe) { + io_uring_prep_read(sqe, static_cast(file.index_), buffer.data(), + static_cast(buffer.size()), offset); + sqe->flags |= IOSQE_FIXED_FILE; + io_uring_sqe_set_data(sqe, tag); + }); } absl::Status IoUringBackend::SubmitWrite(FixedFile file, std::span buffer, std::uint64_t offset, IoCompletion* tag) { - io_uring_sqe prepared{}; - auto* sqe = &prepared; - io_uring_prep_write(sqe, static_cast(file.index_), buffer.data(), - static_cast(buffer.size()), offset); - sqe->flags |= IOSQE_FIXED_FILE; - io_uring_sqe_set_data(sqe, tag); - return SubmitOrQueue(prepared); + return SubmitOrQueue([=](io_uring_sqe* sqe) { + io_uring_prep_write(sqe, static_cast(file.index_), buffer.data(), + static_cast(buffer.size()), offset); + sqe->flags |= IOSQE_FIXED_FILE; + io_uring_sqe_set_data(sqe, tag); + }); } absl::Status IoUringBackend::SubmitWriteFixed(FixedFile file, FixedBuffer buffer, std::uint64_t offset, IoCompletion* tag) { - io_uring_sqe prepared{}; - auto* sqe = &prepared; - io_uring_prep_write_fixed(sqe, static_cast(file.index_), buffer.data_, - static_cast(buffer.size_), offset, - buffer.index_); - sqe->flags |= IOSQE_FIXED_FILE; - io_uring_sqe_set_data(sqe, tag); - return SubmitOrQueue(prepared); + return SubmitOrQueue([=](io_uring_sqe* sqe) { + io_uring_prep_write_fixed(sqe, static_cast(file.index_), buffer.data_, + static_cast(buffer.size_), offset, + buffer.index_); + sqe->flags |= IOSQE_FIXED_FILE; + io_uring_sqe_set_data(sqe, tag); + }); } absl::Status IoUringBackend::SubmitFdatasync(FixedFile file, IoCompletion* tag) { - io_uring_sqe prepared{}; - auto* sqe = &prepared; - io_uring_prep_fsync(sqe, static_cast(file.index_), - IORING_FSYNC_DATASYNC); - sqe->flags |= IOSQE_FIXED_FILE; - io_uring_sqe_set_data(sqe, tag); - return SubmitOrQueue(prepared); + return SubmitOrQueue([=](io_uring_sqe* sqe) { + io_uring_prep_fsync(sqe, static_cast(file.index_), + IORING_FSYNC_DATASYNC); + sqe->flags |= IOSQE_FIXED_FILE; + io_uring_sqe_set_data(sqe, tag); + }); } absl::Status IoUringBackend::SubmitTimeout(const __kernel_timespec& timeout, IoCompletion* tag) { - io_uring_sqe prepared{}; - auto* sqe = &prepared; - io_uring_prep_timeout(sqe, const_cast<__kernel_timespec*>(&timeout), 0, 0); - io_uring_sqe_set_data(sqe, tag); - return SubmitOrQueue(prepared); + // The SQE retains the caller's timespec, never a copy in the short-lived + // preparation callback. + auto* timeout_ptr = const_cast<__kernel_timespec*>(&timeout); + return SubmitOrQueue([timeout_ptr, tag](io_uring_sqe* sqe) { + io_uring_prep_timeout(sqe, timeout_ptr, 0, 0); + io_uring_sqe_set_data(sqe, tag); + }); } io_uring_sqe* IoUringBackend::AcquireSqe() { @@ -356,35 +385,6 @@ io_uring_sqe IoUringBackend::PendingIo::descriptor() const noexcept { return std::bit_cast(descriptor_); } -absl::Status IoUringBackend::SubmitOrQueue(const io_uring_sqe& prepared) { - // Older accepted work retains priority. Never submit/reap here: callers - // install their continuation and connection accounting after we return. - if (initialized_ && pending_io_.empty()) [[likely]] { - if (auto* sqe = AcquireSqe()) [[likely]] { - *sqe = prepared; - if (prepared.opcode == IORING_OP_ASYNC_CANCEL) [[unlikely]] { - may_have_unsubmitted_cancels_ = true; - } - return absl::OkStatus(); - } - } - return QueuePrepared(prepared); -} - -// Deque growth and rejection diagnostics must not expand the ordinary SQE -// preparation path. Backlogs can be sustained, so this helper is not cold. -ABSL_ATTRIBUTE_NOINLINE absl::Status IoUringBackend::QueuePrepared( - const io_uring_sqe& prepared) { - if (!initialized_) { - return absl::FailedPreconditionError("backend is not initialized"); - } - if (prepared.opcode == IORING_OP_ASYNC_CANCEL) { - may_have_unsubmitted_cancels_ = true; - } - pending_io_.push_back(PendingIo{prepared}); - return absl::OkStatus(); -} - bool IoUringBackend::DrainPending() { if (pending_io_.empty()) [[likely]] return false; @@ -471,10 +471,14 @@ absl::Status IoUringBackend::CancelByData(void* target, void* completion) { {reinterpret_cast(completion), 0}); return absl::OkStatus(); } - io_uring_sqe prepared{}; - io_uring_prep_cancel(&prepared, target, 0); - io_uring_sqe_set_data(&prepared, completion); - return SubmitOrQueue(prepared); + auto status = SubmitOrQueue([=](io_uring_sqe* sqe) { + io_uring_prep_cancel(sqe, target, 0); + io_uring_sqe_set_data(sqe, completion); + }); + // Submission cannot dispatch completions inline. Mark the accepted cancel + // here so ordinary operations never need to inspect their opcode. + if (status.ok()) may_have_unsubmitted_cancels_ = true; + return status; } void IoUringBackend::CancelUnsubmittedForFd(int fd) noexcept { @@ -569,10 +573,11 @@ bool IoUringBackend::ArmWakePoll() { if (wake_event_fd_ < 0 || wake_poll_armed_) { return wake_event_fd_ >= 0; } - io_uring_sqe prepared{}; - io_uring_prep_poll_add(&prepared, wake_event_fd_, POLLIN); - io_uring_sqe_set_data(&prepared, &kWakePollTag); - if (!SubmitOrQueue(prepared).ok()) return false; + auto status = SubmitOrQueue([this](io_uring_sqe* sqe) { + io_uring_prep_poll_add(sqe, wake_event_fd_, POLLIN); + io_uring_sqe_set_data(sqe, &kWakePollTag); + }); + if (!status.ok()) return false; wake_poll_armed_ = true; return true; } @@ -596,41 +601,38 @@ void IoUringBackend::HandleWakePoll() { absl::Status IoUringBackend::SubmitSend(const RegisteredFile& file, std::span buffer, IoCompletion* tag) { - io_uring_sqe prepared{}; - auto* sqe = &prepared; - io_uring_prep_send( - sqe, file.is_fixed_ ? static_cast(file.fixed_index_) : file.fd_, - buffer.data(), static_cast(buffer.size()), 0); - if (file.is_fixed_) { - sqe->flags |= IOSQE_FIXED_FILE; - } - io_uring_sqe_set_data(sqe, tag); - return SubmitOrQueue(prepared); + return SubmitOrQueue([=](io_uring_sqe* sqe) { + io_uring_prep_send( + sqe, file.is_fixed_ ? static_cast(file.fixed_index_) : file.fd_, + buffer.data(), static_cast(buffer.size()), 0); + if (file.is_fixed_) { + sqe->flags |= IOSQE_FIXED_FILE; + } + io_uring_sqe_set_data(sqe, tag); + }); } absl::Status IoUringBackend::SubmitSendMsg(const RegisteredFile& file, const msghdr* message, IoCompletion* tag) { - io_uring_sqe prepared{}; - auto* sqe = &prepared; - io_uring_prep_sendmsg( - sqe, file.is_fixed_ ? static_cast(file.fixed_index_) : file.fd_, - message, 0); - if (file.is_fixed_) { - sqe->flags |= IOSQE_FIXED_FILE; - } - io_uring_sqe_set_data(sqe, tag); - return SubmitOrQueue(prepared); + return SubmitOrQueue([=](io_uring_sqe* sqe) { + io_uring_prep_sendmsg( + sqe, file.is_fixed_ ? static_cast(file.fixed_index_) : file.fd_, + message, 0); + if (file.is_fixed_) { + sqe->flags |= IOSQE_FIXED_FILE; + } + io_uring_sqe_set_data(sqe, tag); + }); } absl::Status IoUringBackend::SubmitAcceptMultishot(int listen_fd, IoCompletion* tag) { - io_uring_sqe prepared{}; - auto* sqe = &prepared; - io_uring_prep_multishot_accept(sqe, listen_fd, nullptr, nullptr, - SOCK_NONBLOCK | SOCK_CLOEXEC); - io_uring_sqe_set_data(sqe, tag); // plain IoCompletion* — same path as send - return SubmitOrQueue(prepared); + return SubmitOrQueue([=](io_uring_sqe* sqe) { + io_uring_prep_multishot_accept(sqe, listen_fd, nullptr, nullptr, + SOCK_NONBLOCK | SOCK_CLOEXEC); + io_uring_sqe_set_data(sqe, tag); // plain IoCompletion* — same path as send + }); } absl::Status IoUringBackend::SubmitConnect(int fd, const sockaddr* address, @@ -640,11 +642,10 @@ absl::Status IoUringBackend::SubmitConnect(int fd, const sockaddr* address, return absl::Status(absl::StatusCode::kInvalidArgument, "invalid connect request"); } - io_uring_sqe prepared{}; - auto* sqe = &prepared; - io_uring_prep_connect(sqe, fd, address, address_length); - io_uring_sqe_set_data(sqe, tag); - return SubmitOrQueue(prepared); + return SubmitOrQueue([=](io_uring_sqe* sqe) { + io_uring_prep_connect(sqe, fd, address, address_length); + io_uring_sqe_set_data(sqe, tag); + }); } absl::Status IoUringBackend::SubmitCancel(IoCompletion* target) { @@ -669,30 +670,30 @@ absl::Status IoUringBackend::StartRecvMultishot(Connection* connection) { if (!use_multishot && !connection->received_buffers_.empty()) { return absl::OkStatus(); } - io_uring_sqe prepared{}; - auto* sqe = &prepared; - const int fd = connection->file_.is_fixed_ - ? static_cast(connection->file_.fixed_index_) - : connection->file_.fd_; - if (use_multishot) { - io_uring_prep_recv_multishot(sqe, fd, nullptr, 0, 0); - sqe->flags |= IOSQE_BUFFER_SELECT; - sqe->buf_group = MultishotBufferRing::kGroupId; - } else { - if (connection->read_buffer_.empty()) { - connection->read_buffer_.resize(options_.recv_buffer_size_); + // Allocate before acquiring an SQE: preparation must not throw after the + // ring's local tail advances and expose a partially prepared operation. + if (!use_multishot && connection->read_buffer_.empty()) { + connection->read_buffer_.resize(options_.recv_buffer_size_); + } + auto status = SubmitOrQueue([connection, use_multishot](io_uring_sqe* sqe) { + const int fd = connection->file_.is_fixed_ + ? static_cast(connection->file_.fixed_index_) + : connection->file_.fd_; + if (use_multishot) { + io_uring_prep_recv_multishot(sqe, fd, nullptr, 0, 0); + sqe->flags |= IOSQE_BUFFER_SELECT; + sqe->buf_group = MultishotBufferRing::kGroupId; + } else { + io_uring_prep_recv(sqe, fd, connection->read_buffer_.data(), + static_cast(connection->read_buffer_.size()), + 0); } - io_uring_prep_recv(sqe, fd, connection->read_buffer_.data(), - static_cast(connection->read_buffer_.size()), - 0); - } - sqe->ioprio |= IORING_RECVSEND_POLL_FIRST; - if (connection->file_.is_fixed_) { - sqe->flags |= IOSQE_FIXED_FILE; - } - io_uring_sqe_set_data(sqe, EncodeMultishotData(connection)); - - auto status = SubmitOrQueue(prepared); + sqe->ioprio |= IORING_RECVSEND_POLL_FIRST; + if (connection->file_.is_fixed_) { + sqe->flags |= IOSQE_FIXED_FILE; + } + io_uring_sqe_set_data(sqe, EncodeMultishotData(connection)); + }); if (!status.ok()) return status; connection->recv_armed_ = true; connection->inflight_ops_ += 1; @@ -720,16 +721,15 @@ absl::Status IoUringBackend::StartPeerDisconnectPoll(Connection* connection) { connection->closing_) { return absl::OkStatus(); } - io_uring_sqe prepared{}; - auto* sqe = &prepared; - const int fd = connection->file_.is_fixed_ - ? static_cast(connection->file_.fixed_index_) - : connection->file_.fd_; - io_uring_prep_poll_add(sqe, fd, - static_cast(POLLERR | POLLHUP | POLLRDHUP)); - if (connection->file_.is_fixed_) sqe->flags |= IOSQE_FIXED_FILE; - io_uring_sqe_set_data(sqe, EncodePeerDisconnectData(connection)); - auto status = SubmitOrQueue(prepared); + auto status = SubmitOrQueue([connection](io_uring_sqe* sqe) { + const int fd = connection->file_.is_fixed_ + ? static_cast(connection->file_.fixed_index_) + : connection->file_.fd_; + io_uring_prep_poll_add( + sqe, fd, static_cast(POLLERR | POLLHUP | POLLRDHUP)); + if (connection->file_.is_fixed_) sqe->flags |= IOSQE_FIXED_FILE; + io_uring_sqe_set_data(sqe, EncodePeerDisconnectData(connection)); + }); if (!status.ok()) return status; connection->peer_disconnect_poll_armed_ = true; connection->peer_disconnect_poll_cancel_requested_ = false; @@ -982,11 +982,10 @@ absl::Status IoUringBackend::SubmitPoll(int fd, unsigned events, IoCompletion* tag) { if (fd < 0 || tag == nullptr) return absl::InvalidArgumentError("invalid poll request"); - io_uring_sqe prepared{}; - auto* sqe = &prepared; - io_uring_prep_poll_add(sqe, fd, events); - io_uring_sqe_set_data(sqe, tag); - return SubmitOrQueue(prepared); + return SubmitOrQueue([=](io_uring_sqe* sqe) { + io_uring_prep_poll_add(sqe, fd, events); + io_uring_sqe_set_data(sqe, tag); + }); } bool IoUringBackend::Wait(int timeout_ms) { @@ -1025,12 +1024,12 @@ bool IoUringBackend::Wait(int timeout_ms) { } void IoUringBackend::WakeRemote(int peer_ring_fd) noexcept { - io_uring_sqe prepared{}; - io_uring_prep_msg_ring(&prepared, peer_ring_fd, 0, - reinterpret_cast(&kCrossCoreWakeTag), - 0); - io_uring_sqe_set_data(&prepared, &kCrossCoreWakeTag); - (void)SubmitOrQueue(prepared); + (void)SubmitOrQueue([peer_ring_fd](io_uring_sqe* sqe) { + io_uring_prep_msg_ring(sqe, peer_ring_fd, 0, + reinterpret_cast(&kCrossCoreWakeTag), + 0); + io_uring_sqe_set_data(sqe, &kCrossCoreWakeTag); + }); } void IoUringBackend::WakeSelf() noexcept { diff --git a/tests/io_uring_backpressure_check.cpp b/tests/io_uring_backpressure_check.cpp index c6a1f77..1c0d060 100644 --- a/tests/io_uring_backpressure_check.cpp +++ b/tests/io_uring_backpressure_check.cpp @@ -83,11 +83,22 @@ void CheckSendsAndWake() { constexpr std::size_t kCount = 64; std::array tags; std::array payload; + std::array vectors; + std::array messages{}; + // Submit the initial wake poll so both SQ slots are available to the burst. + CHECK(source.backend.Submit().ok()); for (std::size_t i = 0; i < kCount; ++i) { payload[i] = std::byte(i); - CHECK(source.backend - .SubmitSend({.fd_ = sockets[0]}, std::span(payload).subspan(i, 1), - &tags[i]) + vectors[i] = {.iov_base = &payload[i], .iov_len = 1}; + messages[i].msg_iov = &vectors[i]; + messages[i].msg_iovlen = 1; + // Exercise both preparation callbacks in the real SQ and the pending FIFO. + // Indirect sendmsg arguments remain caller-owned through completion. + CHECK((i % 2 == 0 ? source.backend.SubmitSend( + {.fd_ = sockets[0]}, + std::span(payload).subspan(i, 1), &tags[i]) + : source.backend.SubmitSendMsg({.fd_ = sockets[0]}, + &messages[i], &tags[i])) .ok()); } source.backend.WakeRemote(peer.backend.WakeHandle());