Queue io_uring and SPDK submissions under I/O backpressure

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"

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
@@ -113,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

docs/architecture/io.md

@@ -38,15 +38,37 @@ 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.
-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

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,

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

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 <arpa/inet.h>
 #include <sys/socket.h>
+#include <unistd.h>
 
 #include <array>
 #include <atomic>
@@ -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<absl::Status> Serve(bycorf::TcpStream) override {
+    co_return absl::OkStatus();
+  }
+};
+
+bycorf::Task<absl::Status> RunServiceAndNotify(
+    StopCheckService& service, bycorf::Worker& worker,
+    bycorf::ServiceContext ctx, std::promise<absl::Status>& finished) {
+  auto status = co_await service.Run(worker, ctx);
+  finished.set_value(status);
+  co_return status;
+}
+
+bycorf::Task<absl::Status> NotifyServiceStarted(
+    std::promise<absl::Status>& 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<sockaddr*>(&address), length) == 0 &&
+      getsockname(reservation, reinterpret_cast<sockaddr*>(&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<absl::Status> started, finished;
+  auto ready = started.get_future();
+  auto stopped = finished.get_future();
+  std::promise<void> 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<std::string, 1> 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<void> 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::milliseconds>(
              std::chrono::steady_clock::now() - start)
@@ -473,6 +579,8 @@ bycorf::Task<absl::Status> 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,

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