Handle transient SPDK qpair descriptor exhaustion with deferred submissions

.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

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)

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
 

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

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,13 +73,17 @@ 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;
   SpdkStorageBackend(const SpdkStorageBackend&) = delete;
   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;

include/bycorf/runtime/worker.h

@@ -396,6 +396,12 @@ class Worker {
   void SpawnBackground(Task<absl::Status> 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;

src/io/spdk_storage.cpp

@@ -376,7 +376,9 @@ void SpdkStorageBackend::CompleteAsync(void* context,
   auto* request = static_cast<AsyncRequest*>(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<int>(bytes);
-  const std::uint64_t lba = offset / sector;
-  const std::uint32_t count = static_cast<std::uint32_t>(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<std::uint32_t>(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<SpdkDevice*>(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<std::pair<IoCompletion*, int>> completions;
@@ -687,6 +756,9 @@ SpdkPollResult SpdkStorageBackend::Poll(unsigned max_completions) {
       result.did_work_ = true;
       result.completions_ += static_cast<std::uint32_t>(completed);
     }
+    if (channel.pending_head_ != nullptr) [[unlikely]] {
+      RetryDeferred(channel, max_completions, result);
+    }
     next_poll_channel_ = (index + 1) % channel_count;
   }
   return result;