perf(io): avoid fixed-buffer copies during submission

include/bycorf/io/io_uring_backend.h

@@ -122,13 +122,15 @@ class IoUringBackend {
   absl::Status SubmitOpenDirect(std::string_view path, int flags, mode_t mode,
                                 FixedFile file, IoCompletion* tag);
   absl::Status SubmitCloseDirect(FixedFile file, IoCompletion* tag);
-  absl::Status SubmitReadFixed(FixedFile file, FixedBuffer buffer,
+  // Fixed-buffer submissions borrow the descriptor only for this call. Both
+  // the SQ and the pending FIFO retain prepared values, never its address.
+  absl::Status SubmitReadFixed(FixedFile file, const FixedBuffer& buffer,
                                std::uint64_t offset, IoCompletion* tag);
   absl::Status SubmitRead(FixedFile file, std::span<std::byte> buffer,
                           std::uint64_t offset, IoCompletion* tag);
   absl::Status SubmitWrite(FixedFile file, std::span<const std::byte> buffer,
                            std::uint64_t offset, IoCompletion* tag);
-  absl::Status SubmitWriteFixed(FixedFile file, FixedBuffer buffer,
+  absl::Status SubmitWriteFixed(FixedFile file, const FixedBuffer& buffer,
                                 std::uint64_t offset, IoCompletion* tag);
   absl::Status SubmitFdatasync(FixedFile file, IoCompletion* tag);
   absl::Status SubmitTimeout(const __kernel_timespec& timeout,

include/bycorf/io/spdk_storage.h

@@ -124,13 +124,15 @@ class SpdkBackend {
   absl::Status SubmitOpenDirect(std::string_view path, int flags, mode_t mode,
                                 FixedFile file, IoCompletion* tag);
   absl::Status SubmitCloseDirect(FixedFile file, IoCompletion* tag);
-  absl::Status SubmitReadFixed(FixedFile file, FixedBuffer buffer,
+  // Fixed-buffer submissions borrow the descriptor only for this call;
+  // deferred requests own their buffer address, length and device offset.
+  absl::Status SubmitReadFixed(FixedFile file, const FixedBuffer& buffer,
                                std::uint64_t offset, IoCompletion* tag);
   absl::Status SubmitRead(FixedFile file, std::span<std::byte> buffer,
                           std::uint64_t offset, IoCompletion* tag);
   absl::Status SubmitWrite(FixedFile file, std::span<const std::byte> buffer,
                            std::uint64_t offset, IoCompletion* tag);
-  absl::Status SubmitWriteFixed(FixedFile file, FixedBuffer buffer,
+  absl::Status SubmitWriteFixed(FixedFile file, const FixedBuffer& buffer,
                                 std::uint64_t offset, IoCompletion* tag);
   absl::Status SubmitFdatasync(FixedFile file, IoCompletion* tag);
 

include/bycorf/io/storage_backend.h

@@ -120,7 +120,9 @@ class StorageBackend {
 #endif
     return Kernel().SubmitCloseDirect(file, tag);
   }
-  absl::Status SubmitReadFixed(FixedFile file, FixedBuffer buffer,
+  // Forward the descriptor without copying it. Neither backend retains its
+  // address after submission, including when the request must be queued.
+  absl::Status SubmitReadFixed(FixedFile file, const FixedBuffer& buffer,
                                std::uint64_t offset, IoCompletion* tag) {
 #if BYCORF_KERNEL_BYPASS
     if (spdk_.IsInitialized()) {
@@ -147,7 +149,7 @@ class StorageBackend {
 #endif
     return Kernel().SubmitWrite(file, buffer, offset, tag);
   }
-  absl::Status SubmitWriteFixed(FixedFile file, FixedBuffer buffer,
+  absl::Status SubmitWriteFixed(FixedFile file, const FixedBuffer& buffer,
                                 std::uint64_t offset, IoCompletion* tag) {
 #if BYCORF_KERNEL_BYPASS
     if (spdk_.IsInitialized()) {

include/bycorf/runtime/worker.h

@@ -303,7 +303,10 @@ class Worker : private IoUringBackend,
   absl::Status SubmitCloseDirect(FixedFile file, IoCompletion* tag) {
     return Storage::SubmitCloseDirect(file, tag);
   }
-  absl::Status SubmitReadFixed(FixedFile file, FixedBuffer buffer,
+  // Borrow only the descriptor during submission to avoid copying it through
+  // each facade. Backends snapshot queued arguments before returning; the
+  // referenced data and registration must still survive until completion.
+  absl::Status SubmitReadFixed(FixedFile file, const FixedBuffer& buffer,
                                std::uint64_t offset, IoCompletion* tag) {
     return Storage::SubmitReadFixed(file, buffer, offset, tag);
   }
@@ -320,7 +323,7 @@ class Worker : private IoUringBackend,
     }
     return status;
   }
-  absl::Status SubmitWriteFixed(FixedFile file, FixedBuffer buffer,
+  absl::Status SubmitWriteFixed(FixedFile file, const FixedBuffer& buffer,
                                 std::uint64_t offset, IoCompletion* tag) {
     auto status = Storage::SubmitWriteFixed(file, buffer, offset, tag);
     if (status.ok() && file.index_ < storage_file_io_stats_.size()) {

src/io/io_uring_backend.cpp

@@ -301,7 +301,8 @@ absl::Status IoUringBackend::SubmitCloseDirect(FixedFile file,
   });
 }
 
-absl::Status IoUringBackend::SubmitReadFixed(FixedFile file, FixedBuffer buffer,
+absl::Status IoUringBackend::SubmitReadFixed(FixedFile file,
+                                             const FixedBuffer& buffer,
                                              std::uint64_t offset,
                                              IoCompletion* tag) {
   return SubmitOrQueue([=](io_uring_sqe* sqe) {
@@ -338,7 +339,7 @@ absl::Status IoUringBackend::SubmitWrite(FixedFile file,
 }
 
 absl::Status IoUringBackend::SubmitWriteFixed(FixedFile file,
-                                              FixedBuffer buffer,
+                                              const FixedBuffer& buffer,
                                               std::uint64_t offset,
                                               IoCompletion* tag) {
   return SubmitOrQueue([=](io_uring_sqe* sqe) {

src/io/spdk_storage.cpp

@@ -593,7 +593,8 @@ absl::Status SpdkBackend::SubmitIo(FixedFile file, void* buffer,
   return SubmitOrDefer(request, device->ns_, opened->channel_index_);
 }
 
-absl::Status SpdkBackend::SubmitReadFixed(FixedFile file, FixedBuffer buffer,
+absl::Status SpdkBackend::SubmitReadFixed(FixedFile file,
+                                          const FixedBuffer& buffer,
                                           std::uint64_t offset,
                                           IoCompletion* tag) {
   return SubmitIo(file, buffer.data_, buffer.size_, offset, false, tag);
@@ -612,7 +613,8 @@ absl::Status SpdkBackend::SubmitWrite(FixedFile file,
                   offset, true, tag);
 }
 
-absl::Status SpdkBackend::SubmitWriteFixed(FixedFile file, FixedBuffer buffer,
+absl::Status SpdkBackend::SubmitWriteFixed(FixedFile file,
+                                           const FixedBuffer& buffer,
                                            std::uint64_t offset,
                                            IoCompletion* tag) {
   return SubmitIo(file, buffer.data_, buffer.size_, offset, true, tag);

tests/io_uring_backpressure_test.cpp

@@ -400,6 +400,65 @@ TEST(IoUringBackpressure, StorageReads) {
   std::cout << "PASS queued fixed-file reads preserve buffers and offsets\n";
 }
 
+TEST(IoUringBackpressure, FixedBufferDescriptorLifetime) {
+  Fixture fixture;
+  ASSERT_NO_FATAL_FAILURE(fixture.Init());
+  char path[] = "/tmp/bycorf-fixed-buffer-XXXXXX";
+  const int fd = mkstemp(path);
+  ASSERT_GE(fd, 0);
+  ASSERT_TRUE(fixture.backend.RegisterFixedFiles(1).ok());
+  Completion opened;
+  ASSERT_TRUE(
+      fixture.backend.SubmitOpenDirect(path, O_RDWR, 0, {0}, &opened).ok());
+  Drain(fixture.backend, [&] { return opened.calls != 0; });
+  ASSERT_EQ(opened.result, 0);
+  ASSERT_EQ(unlink(path), 0);
+  ASSERT_EQ(close(fd), 0);
+
+  std::array<std::byte, 64> payload, received{};
+  for (std::size_t i = 0; i < payload.size(); ++i) payload[i] = std::byte(i);
+  const std::array<iovec, 2> regions{
+      {{payload.data(), payload.size()}, {received.data(), received.size()}}};
+  ASSERT_TRUE(fixture.backend.RegisterBuffers(regions).ok());
+  std::array<Completion, 16> writes, reads;
+  bycorf::FixedBuffer descriptor;
+  for (std::size_t i = 0; i < writes.size(); ++i) {
+    descriptor = {.data_ = payload.data() + i * 4, .size_ = 4, .index_ = 0};
+    ASSERT_TRUE(
+        fixture.backend.SubmitWriteFixed({0}, descriptor, i * 4, &writes[i])
+            .ok());
+  }
+  // Most requests are queued behind the two-entry SQ. Reusing and clearing
+  // the descriptor must not change their saved buffer, length or index.
+  descriptor = {};
+  for (const auto& tag : writes) ASSERT_EQ(tag.calls, 0u);
+  Drain(fixture.backend, [&] {
+    for (const auto& tag : writes)
+      if (tag.calls == 0) return false;
+    return true;
+  });
+  for (const auto& tag : writes) ASSERT_EQ(tag.result, 4);
+  for (std::size_t i = 0; i < reads.size(); ++i) {
+    descriptor = {.data_ = received.data() + i * 4, .size_ = 4, .index_ = 1};
+    ASSERT_TRUE(
+        fixture.backend.SubmitReadFixed({0}, descriptor, i * 4, &reads[i])
+            .ok());
+  }
+  descriptor = {};
+  for (const auto& tag : reads) ASSERT_EQ(tag.calls, 0u);
+  Drain(fixture.backend, [&] {
+    for (const auto& tag : reads)
+      if (tag.calls == 0) return false;
+    return true;
+  });
+  for (const auto& tag : reads) ASSERT_EQ(tag.result, 4);
+  ASSERT_EQ(received, payload);
+  Completion closed;
+  ASSERT_TRUE(fixture.backend.SubmitCloseDirect({0}, &closed).ok());
+  Drain(fixture.backend, [&] { return closed.calls != 0; });
+  ASSERT_EQ(closed.result, 0);
+}
+
 // Both facades must submit to the same ring: the network event loop also
 // delivers storage completions, and stopping either facade must leave the
 // worker-owned ring available until the worker shuts it down.

tests/spdk_backpressure_stub_test.cpp

@@ -307,6 +307,40 @@ void CheckQueues(bycorf::Worker& worker, std::span<std::byte> buffer) {
       << "PASS independent qpairs, FIFO retries, read/write/flush metadata\n";
 }
 
+void CheckFixedBufferDescriptorLifetime(bycorf::Worker& worker,
+                                        std::span<std::byte> buffer) {
+  auto& queue = *qpairs[0];
+  Completion first, write, read;
+  bycorf::FixedBuffer descriptor{.data_ = buffer.data(),
+                                 .size_ = buffer.size()};
+  CHECK(worker.SubmitReadFixed({0}, descriptor, 0, &first).ok());
+  descriptor = {.data_ = buffer.data() + 512, .size_ = 1024};
+  CHECK(worker.SubmitWriteFixed({1}, descriptor, 1024, &write).ok());
+  descriptor = {.data_ = buffer.data() + 2048, .size_ = 512};
+  CHECK(worker.SubmitReadFixed({0}, descriptor, 2048, &read).ok());
+  // The qpair admits only the first request. Deferred read/write arguments
+  // must be independent of the caller's reused descriptor before any poll.
+  descriptor = {};
+  CHECK(queue.accepted.size() == 1);
+  CHECK(first.calls + write.calls + read.calls == 0);
+  queue.completion_tokens = 3;
+  Drain(worker, [&] { return read.calls == 1; });
+  CHECK(first.calls == 1 && first.result == static_cast<int>(buffer.size()));
+  CHECK(write.calls == 1 && write.result == 1024);
+  CHECK(read.result == 512);
+  CHECK(queue.accepted.size() == 3);
+  const auto& saved_write = queue.accepted[1];
+  CHECK(saved_write.operation == Operation::kWrite);
+  CHECK(saved_write.context == &write);
+  CHECK(saved_write.buffer == buffer.data() + 512);
+  CHECK(saved_write.lba == 2 && saved_write.count == 2);
+  const auto& saved_read = queue.accepted[2];
+  CHECK(saved_read.operation == Operation::kRead);
+  CHECK(saved_read.context == &read);
+  CHECK(saved_read.buffer == buffer.data() + 2048);
+  CHECK(saved_read.lba == 4 && saved_read.count == 1);
+}
+
 void CheckErrors(bycorf::Worker& worker, std::span<std::byte> buffer) {
   auto& queue = *qpairs[0];
   Completion oversized;
@@ -680,6 +714,15 @@ TEST(SpdkBackpressureDeathTest, Queues) {
       testing::ExitedWithCode(0), "");
 }
 
+TEST(SpdkBackpressureDeathTest, FixedBufferDescriptorLifetime) {
+  ASSERT_EXIT(
+      {
+        RunCase(CheckFixedBufferDescriptorLifetime);
+        std::exit(0);
+      },
+      testing::ExitedWithCode(0), "");
+}
+
 TEST(SpdkBackpressureDeathTest, Errors) {
   ASSERT_EXIT(
       {