diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c0fc1c3..3a10f91 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -119,9 +119,10 @@ jobs: with: persist-credentials: false - name: Initialize runtime dependencies - # Software coverage remains independent of hardware source trees. The - # bypass half compiles the complete backend set; CTest still selects - # io_uring and hardware smoke tests remain separate. + # Non-bypass jobs need only the kernel runtime dependencies. The + # bypass half compiles the complete backend set; CTest covers SPDK + # backpressure with device stubs and otherwise selects io_uring. + # Hardware smoke tests remain separate. run: | git submodule update --init --depth 1 third_party/abseil third_party/liburing if [[ '${{ matrix.bypass }}' == ON ]]; then diff --git a/CMakeLists.txt b/CMakeLists.txt index 3b44e3b..2970445 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -443,11 +443,39 @@ if (BYCORF_BUILD_EXAMPLES OR BYCORF_BUILD_TESTS) endif() if (BYCORF_BUILD_TESTS) + if (BYCORF_KERNEL_BYPASS) + # Explicit hardware paths/allowlists are required; do not run in ctest. + add_executable(bycorf_spdk_backpressure_check tests/spdk_backpressure_check.cpp) + target_link_libraries(bycorf_spdk_backpressure_check PRIVATE bycorf::core) + + # Run the real backend against link-time NVMe stubs in hosted CI. Keep + # device emulation out of the production request path and use SPDK's own + # headers so API changes also type-check the stub signatures. + add_executable(bycorf_spdk_backpressure_stub_check + tests/spdk_backpressure_stub_check.cpp) + target_link_libraries(bycorf_spdk_backpressure_stub_check PRIVATE bycorf::core) + target_include_directories(bycorf_spdk_backpressure_stub_check PRIVATE + "${BYCORF_SPDK_SOURCE_DIR}/include") + foreach(symbol IN ITEMS + spdk_env_init spdk_nvme_probe spdk_nvme_ctrlr_get_ns + spdk_nvme_ns_is_active spdk_nvme_ns_get_sector_size spdk_nvme_ns_get_size + spdk_nvme_ctrlr_get_data spdk_nvme_ctrlr_alloc_io_qpair + spdk_nvme_ctrlr_free_io_qpair spdk_nvme_qpair_get_num_outstanding_reqs + spdk_nvme_qpair_process_completions spdk_nvme_ns_cmd_read + spdk_nvme_ns_cmd_write spdk_nvme_ns_cmd_flush) + target_link_options(bycorf_spdk_backpressure_stub_check PRIVATE + "LINKER:--wrap=${symbol}") + endforeach() + endif() enable_testing() add_test(NAME bycorf_task_transfer_check COMMAND bycorf_task_transfer_check) set_tests_properties(bycorf_task_transfer_check PROPERTIES TIMEOUT 30 LABELS "runtime;task") if (BYCORF_KERNEL_BYPASS) + add_test(NAME bycorf_spdk_backpressure_stub_check + COMMAND bycorf_spdk_backpressure_stub_check) + set_tests_properties(bycorf_spdk_backpressure_stub_check PROPERTIES + TIMEOUT 30 LABELS "runtime;storage;spdk;stub") add_executable(bycorf_freebsd_local_connect_check tests/freebsd_local_connect_check.cpp) target_link_libraries(bycorf_freebsd_local_connect_check PRIVATE bycorf_freebsd) diff --git a/README.md b/README.md index d471933..1655633 100644 --- a/README.md +++ b/README.md @@ -91,8 +91,8 @@ container policy, and enough locked-memory allowance for registered buffers. If startup fails, check the host's `kernel.io_uring_disabled` setting and the shell's `ulimit -l`; compilation alone does not verify these runtime conditions. -CTest registers four software regressions. The Task check has a 30-second -timeout; the other checks have 90-second timeouts: +CTest always registers the following software regressions. The Task check has +a 30-second timeout; the other checks below have 90-second timeouts: | Test | Coverage | |---|---| @@ -107,6 +107,22 @@ immediately, since that exercises connection failure instead of the deadline. These software tests also run in a bypass-capable build with the default kernel/io_uring selection, without activating EAL or accessing devices. +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: + +```bash +ctest --test-dir build -R '^bycorf_spdk_backpressure_stub_check$' --output-on-failure +``` + +`bycorf_spdk_backpressure_check` is a separate read-only hardware check. It is +built in bypass test builds but excluded from CTest; running it requires +explicit SPDK device paths and a configured device allowlist. + To build tests without the example applications: ```bash @@ -224,15 +240,17 @@ ignored; imported sources retain their upstream formatting and notices. The [CI workflow](.github/workflows/ci.yml) runs on pull requests, pushes to `main`, and manual dispatch. One job checks all maintained sources using the -pinned clang-format hook. Four independent jobs build the core library, RPC +pinned clang-format hook. Eight independent jobs build the core library, RPC library, examples, and tests with GCC 13 and Clang 18 in Debug, then run CTest natively with each compiler on AMD64 (`ubuntu-24.04`) and ARM64 -(`ubuntu-24.04-arm`). The workflow enables -io_uring and raises the test shell's memlock limit on its disposable runners. +(`ubuntu-24.04-arm`), with `BYCORF_KERNEL_BYPASS` both `OFF` and `ON`. The +workflow enables io_uring and raises the test shell's memlock limit on its +disposable runners. CTest must discover at least one test. Its JUnit report and detailed logs are -uploaded as per-architecture/compiler artifacts retained for seven days. Hosted CI -sets `BYCORF_KERNEL_BYPASS=OFF`; DPDK/SPDK hardware testing requires a separate host. +uploaded as per-architecture/compiler/bypass artifacts retained for seven days. +The bypass jobs run the SPDK stub regression; DPDK/SPDK hardware testing +requires a separate host. ## License diff --git a/docs/architecture/io.md b/docs/architecture/io.md index be9d06f..057afe1 100644 --- a/docs/architecture/io.md +++ b/docs/architecture/io.md @@ -39,6 +39,14 @@ selected backend. The worker drives completion polling and batching, and does not create a second ring or worker pool for either accelerator. 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. 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/spdk_storage.h b/include/bycorf/io/spdk_storage.h index 3137352..4e42439 100644 --- a/include/bycorf/io/spdk_storage.h +++ b/include/bycorf/io/spdk_storage.h @@ -34,6 +34,7 @@ #include "bycorf/io/storage.h" struct spdk_nvme_qpair; +struct spdk_nvme_ns; struct spdk_nvme_cpl; namespace bycorf { @@ -72,6 +73,9 @@ void FreeStorageBuffer(void* buffer, std::size_t alignment) noexcept; #if BYCORF_KERNEL_BYPASS +// 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. class SpdkStorageBackend { public: SpdkStorageBackend() = default; @@ -79,6 +83,7 @@ class SpdkStorageBackend { SpdkStorageBackend& operator=(const SpdkStorageBackend&) = delete; ~SpdkStorageBackend(); + // Initialize only the storage backend embedded in worker. absl::Status Init(Worker* worker); void Shutdown(); @@ -97,6 +102,8 @@ class SpdkStorageBackend { std::uint64_t offset, IoCompletion* tag); absl::Status SubmitFdatasync(FixedFile file, IoCompletion* tag); + // Deliver completions on the owning worker, with its worker TLS bound. + // This also applies when Shutdown drains outstanding requests. SpdkPollResult Poll(unsigned max_completions = 0); bool HasOutstanding() const noexcept { return outstanding_ != 0 || !pending_completions_.empty(); @@ -105,10 +112,21 @@ class SpdkStorageBackend { private: struct AsyncRequest { - SpdkStorageBackend* backend_ = nullptr; IoCompletion* tag_ = nullptr; - int success_result_ = 0; 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) + // remains outstanding, so the namespace and qpair stay valid. + FixedFile file_; + enum class Operation { + kRead, + kWrite, + kFlush + } operation_ = Operation::kRead; }; struct OpenFile { @@ -120,10 +138,20 @@ class SpdkStorageBackend { void* controller_ = nullptr; spdk_nvme_qpair* qpair_ = nullptr; unsigned open_files_ = 0; + AsyncRequest* pending_head_ = nullptr; + AsyncRequest* pending_tail_ = nullptr; }; 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, + spdk_nvme_qpair* qpair); + absl::Status SubmitOrDefer(AsyncRequest* request, spdk_nvme_ns* ns, + spdk_nvme_qpair* qpair); + void RetryDeferred(ControllerChannel& channel, unsigned limit, + SpdkPollResult& result); OpenFile* Lookup(FixedFile file); AsyncRequest* AcquireRequest() noexcept; void ReleaseRequest(AsyncRequest* request) noexcept; diff --git a/include/bycorf/runtime/worker.h b/include/bycorf/runtime/worker.h index 1fbf7b7..1e51abe 100644 --- a/include/bycorf/runtime/worker.h +++ b/include/bycorf/runtime/worker.h @@ -396,6 +396,12 @@ class Worker { void SpawnBackground(Task task); private: +#if BYCORF_KERNEL_BYPASS + // SPDK callbacks resolve this worker's sole storage backend through the + // existing worker TLS instead of storing an owner pointer in every request. + friend class SpdkStorageBackend; +#endif + struct StorageFileIoStats { std::uint64_t submitted_write_bytes_ = 0; std::uint64_t durable_write_bytes_ = 0; diff --git a/src/io/spdk_storage.cpp b/src/io/spdk_storage.cpp index 5470b6c..4ef1060 100644 --- a/src/io/spdk_storage.cpp +++ b/src/io/spdk_storage.cpp @@ -376,7 +376,9 @@ void SpdkStorageBackend::CompleteAsync(void* context, auto* request = static_cast(context); const int result = spdk_nvme_cpl_is_error(completion) ? -EIO : request->success_result_; - SpdkStorageBackend* backend = request->backend_; + Worker* worker = ThisWorker().self_; + assert(worker != nullptr); + SpdkStorageBackend* backend = &worker->storage_backend_; IoCompletion* tag = request->tag_; backend->CompleteRequest(tag, result); backend->ReleaseRequest(request); @@ -390,13 +392,14 @@ void SpdkStorageBackend::CompleteRequest(IoCompletion* tag, int result) { } absl::Status SpdkStorageBackend::Init(Worker* worker) { - if (worker == nullptr) { + if (worker == nullptr || &worker->storage_backend_ != this) { return absl::Status(absl::StatusCode::kInvalidArgument, - "SPDK storage backend requires a worker"); + "SPDK storage backend must be owned by its worker"); } worker_ = worker; - // This is deliberately much larger than the registered read-buffer count. - // It covers foreground reads plus writes/flushes without a hot-path malloc. + // 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_) { @@ -472,7 +475,6 @@ SpdkStorageBackend::AcquireRequest() noexcept { } void SpdkStorageBackend::ReleaseRequest(AsyncRequest* request) noexcept { - request->backend_ = nullptr; request->tag_ = nullptr; request->success_result_ = 0; request->next_ = free_requests_; @@ -569,22 +571,17 @@ absl::Status SpdkStorageBackend::SubmitIo(FixedFile file, void* buffer, return absl::Status(absl::StatusCode::kResourceExhausted, "SPDK request pool exhausted"); } - request->backend_ = this; request->tag_ = tag; request->success_result_ = static_cast(bytes); - const std::uint64_t lba = offset / sector; - const std::uint32_t count = static_cast(bytes / sector); - int rc = write ? spdk_nvme_ns_cmd_write(device->ns_, opened->qpair_, buffer, - lba, count, CompleteAsync, request, 0) - : spdk_nvme_ns_cmd_read(device->ns_, opened->qpair_, buffer, - lba, count, CompleteAsync, request, 0); - if (rc != 0) { - ReleaseRequest(request); - return absl::Status(absl::StatusCode::kUnavailable, - "SPDK I/O submission failed"); - } - ++outstanding_; - return absl::OkStatus(); + 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; + // 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_); } absl::Status SpdkStorageBackend::SubmitReadFixed(FixedFile file, @@ -640,20 +637,92 @@ absl::Status SpdkStorageBackend::SubmitFdatasync(FixedFile file, return absl::Status(absl::StatusCode::kResourceExhausted, "SPDK request pool exhausted"); } - request->backend_ = this; request->tag_ = tag; request->success_result_ = 0; - if (spdk_nvme_ns_cmd_flush(device->ns_, opened->qpair_, CompleteAsync, - request) != 0) { - ReleaseRequest(request); - return absl::Status(absl::StatusCode::kUnavailable, - "SPDK flush submission failed"); + request->file_ = file; + request->operation_ = AsyncRequest::Operation::kFlush; + return SubmitOrDefer(request, device->ns_, opened->qpair_); +} + +int SpdkStorageBackend::TrySubmit(AsyncRequest* 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); + } + return -EINVAL; +} + +absl::Status SpdkStorageBackend::SubmitOrDefer(AsyncRequest* request, + spdk_nvme_ns* ns, + spdk_nvme_qpair* 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(); + } + ReleaseRequest(request); + return absl::Status(absl::StatusCode::kUnavailable, + "SPDK I/O submission failed: rc=" + std::to_string(rc)); +} + +void SpdkStorageBackend::RetryDeferred(ControllerChannel& channel, + unsigned limit, SpdkPollResult& result) { + for (unsigned submitted = 0; + channel.pending_head_ != nullptr && (limit == 0 || submitted < limit); + ++submitted) { + AsyncRequest* request = channel.pending_head_; + const OpenFile& opened = files_[request->file_.index_]; + assert(opened.qpair_ == channel.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) { + break; + } + channel.pending_head_ = request->next_; + if (channel.pending_head_ == nullptr) channel.pending_tail_ = nullptr; + request->next_ = nullptr; + 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_; + } } - ++outstanding_; - return absl::OkStatus(); } SpdkPollResult SpdkStorageBackend::Poll(unsigned max_completions) { + assert(!initialized_ || ThisWorker().self_ == worker_); SpdkPollResult result; if (!pending_completions_.empty()) { std::vector> completions; @@ -687,6 +756,9 @@ SpdkPollResult SpdkStorageBackend::Poll(unsigned max_completions) { result.did_work_ = true; result.completions_ += static_cast(completed); } + if (channel.pending_head_ != nullptr) [[unlikely]] { + RetryDeferred(channel, max_completions, result); + } next_poll_channel_ = (index + 1) % channel_count; } return result; diff --git a/tests/spdk_backpressure_check.cpp b/tests/spdk_backpressure_check.cpp new file mode 100644 index 0000000..7954717 --- /dev/null +++ b/tests/spdk_backpressure_check.cpp @@ -0,0 +1,162 @@ +/* + * 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 "bycorf/io/spdk_storage.h" +#include "bycorf/runtime/worker.h" + +namespace { +struct Completion : bycorf::IoCompletion { + unsigned calls = 0; + int result = -1; + unsigned* completed_total = nullptr; + void Complete(bycorf::Worker&, int value, unsigned) override { + ++calls; + result = value; + if (completed_total != nullptr) ++*completed_total; + } +}; + +template +void Drain(bycorf::Worker& worker, Done done) { + const auto deadline = + std::chrono::steady_clock::now() + std::chrono::seconds(30); + while (!done()) { + // An empty poll is not a failure; NVMe may complete on a later pass. + worker.RunOnce(false); + if (std::chrono::steady_clock::now() > deadline) { + std::cerr << "FAIL timed out draining accepted I/O\n"; + // Never reclaim DMA buffers while the device may still use them. + std::_Exit(1); + } + } +} +} // namespace + +// Read-only hardware regression: supply one or more explicitly allowed SPDK +// namespaces via argv and the ordinary BYCORF_EAL_ARGS device allowlist. +// Each burst exceeds the default qpair's 512 request descriptors. No polling +// occurs during submission, making queue exhaustion deterministic. +int main(int argc, char** argv) { + if (argc < 2 || argc > 3) { + std::cerr << "usage: spdk_backpressure_check spdk://BDF/1 [spdk://BDF/1]\n"; + return 2; + } + const auto configured = bycorf::ConfigureIoBackends({.spdk_storage = true}); + if (!configured.ok()) { + std::cerr << configured << '\n'; + return 1; + } + for (int i = 1; i < argc; ++i) { + const auto info = bycorf::ProbeSpdkStorage(argv[i]); + if (!info.ok()) { + std::cerr << info.status() << '\n'; + return 1; + } + } + bycorf::ReleaseSpdkStorageMetadataQpairs(); + // Use the real one-backend-per-worker ownership and callback context. + bycorf::CrossCore cross_core(1); + const int wake_fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); + if (wake_fd < 0) return 1; + cross_core.mailbox(0).wake_fd_ = wake_fd; + bycorf::Worker worker; + worker.BindCrossCore(0, &cross_core); + bycorf::SetThisWorker(0, &cross_core, &worker); + const unsigned files = argc - 1; + if (!worker.Init({.spdk_max_completions_per_poll_ = 16}).ok() || + !worker.RegisterFixedFiles(files).ok()) + return 1; + for (unsigned i = 0; i < files; ++i) { + Completion opened; + const auto status = + worker.SubmitOpenDirect(argv[i + 1], O_RDWR, 0, {i}, &opened); + if (!status.ok()) { + std::cerr << status << '\n'; + return 1; + } + if (opened.calls != 0) return 1; + Drain(worker, [&] { return opened.calls != 0; }); + if (opened.calls != 1 || opened.result != 0) return 1; + } + constexpr unsigned per_file = 1024; + constexpr std::size_t bytes = 4096; + const unsigned count = per_file * files; + auto* buffers = static_cast( + bycorf::AllocateStorageBuffer(count * bytes, bytes)); + if (buffers == nullptr) return 1; + bool passed = true; + for (unsigned round = 0; round < 4; ++round) { + std::vector tags(count); + std::vector accepted(count); + unsigned rejected = 0; + unsigned delivered = 0; + for (unsigned i = 0; i < count; ++i) { + tags[i].completed_total = &delivered; + const auto status = worker.SubmitRead( + {i % files}, {buffers + i * bytes, bytes}, 0, &tags[i]); + accepted[i] = status.ok(); + rejected += !status.ok(); + passed &= tags[i].calls == 0; + } + Completion closed; + passed &= !worker.SubmitCloseDirect({0}, &closed).ok(); + Drain(worker, [&] { return delivered == count - rejected; }); + unsigned completed = 0; + for (unsigned i = 0; i < count; ++i) { + if (!accepted[i]) continue; + const bool valid = tags[i].calls == 1 && tags[i].result == bytes && + std::memcmp(buffers + i * bytes, + buffers + (i % files) * bytes, bytes) == 0; + passed &= valid; + completed += valid; + } + passed &= rejected == 0 && completed == count; + std::cout << "round=" << round << " submitted=" << count + << " rejected=" << rejected << " completed=" << completed << '\n'; + } + for (unsigned i = 0; i < files; ++i) { + Completion closed; + passed &= worker.SubmitCloseDirect({i}, &closed).ok(); + Drain(worker, [&] { return closed.calls != 0; }); + passed &= closed.calls == 1 && closed.result == 0; + } + // Shutdown must finish an accepted read under the same worker TLS before + // its caller-owned buffer and completion tag are reclaimed. + Completion reopened; + if (!worker.SubmitOpenDirect(argv[1], O_RDWR, 0, {0}, &reopened).ok()) + return 1; + Drain(worker, [&] { return reopened.calls != 0; }); + passed &= reopened.calls == 1 && reopened.result == 0; + Completion shutdown; + if (!worker.SubmitRead({0}, {buffers, bytes}, 0, &shutdown).ok()) return 1; + passed &= shutdown.calls == 0; + worker.Shutdown(); + passed &= shutdown.calls == 1 && shutdown.result == bytes; + bycorf::FreeStorageBuffer(buffers, bytes); + close(wake_fd); + std::cout << (passed ? "PASS" : "FAIL") << " SPDK request backpressure\n"; + return passed ? 0 : 1; +} diff --git a/tests/spdk_backpressure_stub_check.cpp b/tests/spdk_backpressure_stub_check.cpp new file mode 100644 index 0000000..cab3acc --- /dev/null +++ b/tests/spdk_backpressure_stub_check.cpp @@ -0,0 +1,404 @@ +/* + * 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. + */ + +// 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. +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "bycorf/io/spdk_storage.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) + +enum class Operation { kRead, kWrite, kFlush }; +struct Namespace { + unsigned controller; +}; +struct Controller { + std::array namespaces; +}; +std::array controllers = [] { + std::array result{}; + for (unsigned i = 0; i < result.size(); ++i) { + for (auto& ns : result[i].namespaces) ns.controller = i; + } + return result; +}(); +constexpr std::array paths{ + "spdk://0000:01:00.0/1", "spdk://0000:01:00.0/2", "spdk://0000:02:00.0/1"}; + +struct Command { + Operation operation; + spdk_nvme_ns* ns; + void* buffer; + uint64_t lba; + uint32_t count; + spdk_nvme_cmd_cb callback; + void* context; +}; +struct Qpair { + Controller* controller; + std::deque inflight; + std::vector accepted; + unsigned capacity = 1; + // Completions require explicit tokens, so tests can keep one qpair stalled + // while another makes progress, independently of Worker polling frequency. + unsigned completion_tokens = 0; + unsigned attempts = 0; + int submit_error = 0; + bool completion_error = false; +}; +std::vector qpairs; +unsigned environment_calls = 0; + +Qpair& Queue(spdk_nvme_qpair* qpair) { + auto* queue = reinterpret_cast(qpair); + CHECK(std::find(qpairs.begin(), qpairs.end(), queue) != qpairs.end()); + return *queue; +} + +int Submit(spdk_nvme_qpair* qpair, Command command) { + auto& queue = Queue(qpair); + CHECK(reinterpret_cast(command.ns)->controller == + static_cast(queue.controller - controllers.data())); + ++queue.attempts; + if (queue.submit_error != 0) return queue.submit_error; + if (queue.inflight.size() >= queue.capacity) return -ENOMEM; + queue.inflight.push_back(command); + queue.accepted.push_back(command); + // SPDK does not retain callbacks or payloads when submission is rejected. + return 0; +} +} // namespace + +extern "C" { +int __wrap_spdk_env_init(const spdk_env_opts*) { + ++environment_calls; + return 0; +} +int __wrap_spdk_nvme_probe(const spdk_nvme_transport_id* trid, void* context, + spdk_nvme_probe_cb probe, spdk_nvme_attach_cb attach, + spdk_nvme_remove_cb) { + unsigned index; + if (std::string_view(trid->traddr) == "0000:01:00.0") + index = 0; + else if (std::string_view(trid->traddr) == "0000:02:00.0") + index = 1; + else + return -ENODEV; + spdk_nvme_ctrlr_opts options{}; + CHECK(probe(context, trid, &options)); + attach(context, trid, reinterpret_cast(&controllers[index]), + &options); + return 0; +} +spdk_nvme_ns* __wrap_spdk_nvme_ctrlr_get_ns(spdk_nvme_ctrlr* ctrlr, + uint32_t id) { + if (id < 1 || id > 2) return nullptr; + return reinterpret_cast( + &reinterpret_cast(ctrlr)->namespaces[id - 1]); +} +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; } +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; +} +spdk_nvme_qpair* __wrap_spdk_nvme_ctrlr_alloc_io_qpair( + spdk_nvme_ctrlr* ctrlr, const spdk_nvme_io_qpair_opts*, size_t) { + auto* queue = new Qpair{.controller = reinterpret_cast(ctrlr)}; + qpairs.push_back(queue); + return reinterpret_cast(queue); +} +int __wrap_spdk_nvme_ctrlr_free_io_qpair(spdk_nvme_qpair* qpair) { + auto* queue = &Queue(qpair); + CHECK(queue->inflight.empty()); + std::erase(qpairs, queue); + delete queue; + return 0; +} +uint32_t __wrap_spdk_nvme_qpair_get_num_outstanding_reqs( + spdk_nvme_qpair* qpair) { + return Queue(qpair).inflight.size(); +} +int32_t __wrap_spdk_nvme_qpair_process_completions(spdk_nvme_qpair* qpair, + uint32_t limit) { + auto& queue = Queue(qpair); + unsigned completed = 0; + while (!queue.inflight.empty() && queue.completion_tokens != 0 && + (limit == 0 || completed < limit)) { + auto command = queue.inflight.front(); + queue.inflight.pop_front(); + --queue.completion_tokens; + ++completed; + spdk_nvme_cpl completion{}; + if (queue.completion_error) + completion.status.sc = SPDK_NVME_SC_INTERNAL_DEVICE_ERROR; + command.callback(command.context, &completion); + } + return completed; +} +int __wrap_spdk_nvme_ns_cmd_read(spdk_nvme_ns* ns, spdk_nvme_qpair* qpair, + void* buffer, uint64_t lba, uint32_t count, + spdk_nvme_cmd_cb callback, void* context, + uint32_t flags) { + CHECK(flags == 0); + return Submit(qpair, + {Operation::kRead, ns, buffer, lba, count, callback, context}); +} +int __wrap_spdk_nvme_ns_cmd_write(spdk_nvme_ns* ns, spdk_nvme_qpair* qpair, + void* buffer, uint64_t lba, uint32_t count, + spdk_nvme_cmd_cb callback, void* context, + uint32_t flags) { + CHECK(flags == 0); + return Submit(qpair, + {Operation::kWrite, ns, buffer, lba, count, callback, context}); +} +int __wrap_spdk_nvme_ns_cmd_flush(spdk_nvme_ns* ns, spdk_nvme_qpair* qpair, + spdk_nvme_cmd_cb callback, void* context) { + return Submit(qpair, + {Operation::kFlush, ns, nullptr, 0, 0, callback, context}); +} +} // extern "C" + +// Linker wrapping does not type-check against the SPDK declarations by itself. +#define CHECK_SIGNATURE(name) \ + static_assert(std::is_same_v) +CHECK_SIGNATURE(spdk_env_init); +CHECK_SIGNATURE(spdk_nvme_probe); +CHECK_SIGNATURE(spdk_nvme_ctrlr_get_ns); +CHECK_SIGNATURE(spdk_nvme_ns_is_active); +CHECK_SIGNATURE(spdk_nvme_ns_get_sector_size); +CHECK_SIGNATURE(spdk_nvme_ns_get_size); +CHECK_SIGNATURE(spdk_nvme_ctrlr_get_data); +CHECK_SIGNATURE(spdk_nvme_ctrlr_alloc_io_qpair); +CHECK_SIGNATURE(spdk_nvme_ctrlr_free_io_qpair); +CHECK_SIGNATURE(spdk_nvme_qpair_get_num_outstanding_reqs); +CHECK_SIGNATURE(spdk_nvme_qpair_process_completions); +CHECK_SIGNATURE(spdk_nvme_ns_cmd_read); +CHECK_SIGNATURE(spdk_nvme_ns_cmd_write); +CHECK_SIGNATURE(spdk_nvme_ns_cmd_flush); +#undef CHECK_SIGNATURE + +namespace { +struct Completion : bycorf::IoCompletion { + unsigned calls = 0; + int result = -1; + void Complete(bycorf::Worker& worker, int value, unsigned flags) override { + CHECK(&worker == bycorf::ThisWorker().self_); + CHECK(flags == 0); + CHECK(++calls == 1); + result = value; + } +}; + +template +void Drain(bycorf::Worker& worker, Done done) { + for (unsigned i = 0; i < 20000 && !done(); ++i) worker.RunOnce(false); + CHECK(done()); +} + +void CheckQueues(bycorf::Worker& worker, std::span buffer) { + auto& first = *qpairs[0]; + auto& second = *qpairs[1]; + Completion read, write, flush, other, other_deferred; + CHECK(worker.SubmitRead({0}, buffer, 512, &read).ok()); + CHECK(worker.SubmitWrite({1}, buffer, 1024, &write).ok()); + CHECK(worker.SubmitFdatasync({0}, &flush).ok()); + CHECK(worker.SubmitRead({2}, buffer, 1536, &other).ok()); + CHECK(worker.SubmitRead({2}, buffer, 2048, &other_deferred).ok()); + CHECK(read.calls + write.calls + flush.calls + other.calls + + other_deferred.calls == + 0); + CHECK(first.accepted.size() == 1 && second.accepted.size() == 1); + unsigned attempts = first.attempts; + for (unsigned i = 0; i < 3; ++i) worker.RunOnce(false); + CHECK(first.attempts > attempts); + CHECK(first.accepted.size() == 1); + CHECK(read.calls + write.calls + flush.calls + other.calls + + other_deferred.calls == + 0); + + second.completion_tokens = 2; + Drain(worker, [&] { return other_deferred.calls == 1; }); + CHECK(other.calls == 1 && + other_deferred.result == static_cast(buffer.size())); + CHECK(other.result == static_cast(buffer.size())); + 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].ns == + reinterpret_cast(&controllers[1].namespaces[0])); + CHECK(second.accepted[1].buffer == buffer.data() && + second.accepted[1].lba == 4 && second.accepted[1].count == 8); + + first.completion_tokens = 1; + Drain(worker, [&] { return first.accepted.size() == 2; }); + CHECK(read.calls == 1 && read.result == static_cast(buffer.size())); + CHECK(write.calls == 0 && flush.calls == 0); + const auto& retried = first.accepted[1]; + CHECK(retried.operation == Operation::kWrite); + CHECK(retried.ns == + reinterpret_cast(&controllers[0].namespaces[1])); + CHECK(retried.buffer == buffer.data() && retried.lba == 2 && + retried.count == 8); + first.completion_tokens = 1; + Drain(worker, [&] { return first.accepted.size() == 3; }); + 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].ns == + reinterpret_cast(&controllers[0].namespaces[0])); + first.completion_tokens = 1; + Drain(worker, [&] { return flush.calls == 1; }); + CHECK(flush.result == 0); + std::cout + << "PASS independent qpairs, FIFO retries, read/write/flush metadata\n"; +} + +void CheckErrors(bycorf::Worker& worker, std::span buffer) { + auto& queue = *qpairs[0]; + 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) { + CHECK(worker.SubmitRead({0}, buffer, 0, &rejected).code() == + absl::StatusCode::kUnavailable); + } + CHECK(rejected.calls == 0); + queue.submit_error = 0; + } + for (int error : {-EINVAL, -ENOMEM}) { + Completion accepted, deferred; + CHECK(worker.SubmitRead({0}, buffer, 0, &accepted).ok()); + CHECK(worker.SubmitRead({0}, buffer, 0, &deferred).ok()); + CHECK(accepted.calls + deferred.calls == 0); + queue.submit_error = error; + if (error == -ENOMEM) queue.completion_tokens = 1; + // EINVAL fails the retry even with I/O in flight. ENOMEM fails it once + // the last in-flight completion is gone: neither may strand the request. + Drain(worker, [&] { return deferred.calls == 1; }); + CHECK(deferred.result == error); + CHECK(accepted.calls == (error == -ENOMEM ? 1U : 0U)); + queue.submit_error = 0; + if (accepted.calls == 0) queue.completion_tokens = 1; + Drain(worker, [&] { return accepted.calls == 1; }); + CHECK(accepted.result == static_cast(buffer.size())); + } + Completion failed; + queue.completion_error = true; + CHECK(worker.SubmitRead({0}, buffer, 0, &failed).ok()); + queue.completion_tokens = 1; + Drain(worker, [&] { return failed.calls == 1; }); + CHECK(failed.result == -EIO); + queue.completion_error = false; + std::cout << "PASS immediate errors, retry errors, nonrecoverable ENOMEM, " + "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); + CHECK(!worker.SubmitCloseDirect({0}, &closed).ok()); + CHECK(!worker.RegisterFixedFiles(4).ok()); + qpairs[0]->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()); + } + std::cout << "PASS bounded pool, close protection and complete pool reuse\n"; +} +} // namespace + +int main() { + CHECK(bycorf::ConfigureIoBackends({.spdk_storage = true}).ok()); + for (auto path : paths) CHECK(bycorf::ProbeSpdkStorage(path).ok()); + CHECK(environment_calls == 1); + bycorf::ReleaseSpdkStorageMetadataQpairs(); + CHECK(qpairs.empty()); + bycorf::CrossCore cross_core(1); + const int wake_fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); + CHECK(wake_fd >= 0); + cross_core.mailbox(0).wake_fd_ = wake_fd; + bycorf::Worker worker; + worker.BindCrossCore(0, &cross_core); + bycorf::SetThisWorker(0, &cross_core, &worker); + CHECK(worker.Init({.spdk_max_completions_per_poll_ = 1}).ok()); + CHECK(worker.RegisterFixedFiles(paths.size()).ok()); + std::array opened; + for (unsigned i = 0; i < paths.size(); ++i) { + CHECK(worker.SubmitOpenDirect(paths[i], O_RDWR, 0, {i}, &opened[i]).ok()); + CHECK(opened[i].calls == 0); + } + Drain(worker, [&] { return opened.back().calls == 1; }); + for (auto& tag : opened) CHECK(tag.calls == 1 && tag.result == 0); + CHECK(qpairs.size() == 2); + // Stub commands do not DMA: ordinary memory suffices, with no hugepages, + // device binding, root privileges or PCI discovery required. + std::array buffer{}; + CheckQueues(worker, buffer); + CheckErrors(worker, buffer); + CheckPool(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; + worker.Shutdown(); + CHECK(read.calls == 1 && read.result == 4096); + CHECK(deferred.calls == 1 && deferred.result == 4096); + CHECK(qpairs.empty()); + close(wake_fd); + std::cout << "PASS shutdown drains in-flight and deferred I/O on the owning " + "worker\n"; +}