diff --git a/include/bycorf/io/io_uring_backend.h b/include/bycorf/io/io_uring_backend.h index 4743181..8677d25 100644 --- a/include/bycorf/io/io_uring_backend.h +++ b/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 buffer, std::uint64_t offset, IoCompletion* tag); absl::Status SubmitWrite(FixedFile file, std::span 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, diff --git a/include/bycorf/io/spdk_storage.h b/include/bycorf/io/spdk_storage.h index 9fdbed5..c24d7a3 100644 --- a/include/bycorf/io/spdk_storage.h +++ b/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 buffer, std::uint64_t offset, IoCompletion* tag); absl::Status SubmitWrite(FixedFile file, std::span 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); diff --git a/include/bycorf/io/storage_backend.h b/include/bycorf/io/storage_backend.h index 4c052f7..607cc33 100644 --- a/include/bycorf/io/storage_backend.h +++ b/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()) { diff --git a/include/bycorf/runtime/worker.h b/include/bycorf/runtime/worker.h index 6b3b8c8..a3168e3 100644 --- a/include/bycorf/runtime/worker.h +++ b/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()) { diff --git a/src/io/io_uring_backend.cpp b/src/io/io_uring_backend.cpp index f4dbbb0..21dba19 100644 --- a/src/io/io_uring_backend.cpp +++ b/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) { diff --git a/src/io/spdk_storage.cpp b/src/io/spdk_storage.cpp index b1f888e..427f85f 100644 --- a/src/io/spdk_storage.cpp +++ b/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); diff --git a/tests/io_uring_backpressure_test.cpp b/tests/io_uring_backpressure_test.cpp index 790851c..bb593ec 100644 --- a/tests/io_uring_backpressure_test.cpp +++ b/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 payload, received{}; + for (std::size_t i = 0; i < payload.size(); ++i) payload[i] = std::byte(i); + const std::array regions{ + {{payload.data(), payload.size()}, {received.data(), received.size()}}}; + ASSERT_TRUE(fixture.backend.RegisterBuffers(regions).ok()); + std::array 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. diff --git a/tests/spdk_backpressure_stub_test.cpp b/tests/spdk_backpressure_stub_test.cpp index 1d57df0..a910d77 100644 --- a/tests/spdk_backpressure_stub_test.cpp +++ b/tests/spdk_backpressure_stub_test.cpp @@ -307,6 +307,40 @@ void CheckQueues(bycorf::Worker& worker, std::span buffer) { << "PASS independent qpairs, FIFO retries, read/write/flush metadata\n"; } +void CheckFixedBufferDescriptorLifetime(bycorf::Worker& worker, + std::span 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(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 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( {