From cef5b900dd6ec97159b316cf8e90530b7fd08f53 Mon Sep 17 00:00:00 2001 From: Alexis Jeandet Date: Mon, 2 Nov 2026 10:23:36 +0100 Subject: [PATCH 1/7] feat(io): range_reader concept and local_range_reader Co-Authored-By: Claude Opus 5.5 --- docs/api/io.md | 55 ++++++++++ include/cpp_utils/io/local_range_reader.hpp | 109 ++++++++++++++++++++ include/cpp_utils/io/range_reader.hpp | 50 +++++++++ tests/io/test_local_range_reader.cpp | 69 +++++++++++++ tests/meson.build | 5 + 5 files changed, 288 insertions(+) create mode 100644 include/cpp_utils/io/local_range_reader.hpp create mode 100644 include/cpp_utils/io/range_reader.hpp create mode 100644 tests/io/test_local_range_reader.cpp diff --git a/docs/api/io.md b/docs/api/io.md index e4146cd..e6d90c7 100644 --- a/docs/api/io.md +++ b/docs/api/io.md @@ -359,3 +359,58 @@ writer.offset(); // 6 // file closed automatically when writer is destroyed ``` + +## range_reader + +```cpp +#include +``` + +A source of bytes fetched by ranges, such as an object on S3. It is the transport that +`remote_buffer` reads through. + +```cpp +struct byte_range +{ + std::size_t offset; + std::size_t size; +}; + +template +concept range_reader = std::copy_constructible && requires(const R& r, + std::span ranges, std::span destinations) { + { r.size() } -> std::convertible_to; + r.read_ranges(ranges, destinations); +}; +``` + +- `size()` — the total size of the source, in bytes. +- `read_ranges(ranges, destinations)` — fills `destinations[i]` with exactly the bytes of + `ranges[i]`, or throws. A batch never succeeds partially. Batching lets a transport fetch + the ranges concurrently; it does not merge them. + +## local_range_reader + +```cpp +#include +``` + +A `range_reader` over a local file that logs every call. It is meant for tests and for +measuring what a remote reader would be asked for. Copies share the file and the log, so a +copy handed to another object still reports to the original. + +- `local_range_reader(path)` — maps the file. +- `batches()` — a snapshot of every `read_ranges` call, one vector of ranges per call. +- `bytes_served()` — the total bytes handed out. +- `caller_threads()` — the thread id of each call, in call order. +- A range past the end throws `std::out_of_range`. + +```cpp +cpp_utils::io::local_range_reader reader { "data.bin" }; +const cpp_utils::io::byte_range ranges[] { { 0, 4 }, { 8, 4 } }; +char a[4], b[4]; +char* const destinations[] { a, b }; +reader.read_ranges(ranges, destinations); +reader.batches().size(); // 1 +reader.bytes_served(); // 8 +``` diff --git a/include/cpp_utils/io/local_range_reader.hpp b/include/cpp_utils/io/local_range_reader.hpp new file mode 100644 index 0000000..539e9da --- /dev/null +++ b/include/cpp_utils/io/local_range_reader.hpp @@ -0,0 +1,109 @@ +/*------------------------------------------------------------------------------ +-- The MIT License (MIT) +-- +-- Copyright © 2024, Laboratory of Plasma Physics- CNRS +-- +-- Permission is hereby granted, free of charge, to any person obtaining a copy +-- of this software and associated documentation files (the “Software”), to deal +-- in the Software without restriction, including without limitation the rights +-- to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies +-- of the Software, and to permit persons to whom the Software is furnished to do +-- so, subject to the following conditions: +-- +-- The above copyright notice and this permission notice shall be included in all +-- copies or substantial portions of the Software. +-- +-- THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, +-- INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +-- PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +-- HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +-- OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE +-- SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +-------------------------------------------------------------------------------*/ +/*-- Author : Alexis Jeandet +-- Mail : alexis.jeandet@member.fsf.org +----------------------------------------------------------------------------*/ +#pragma once + +#include "memory_mapped_file.hpp" +#include "range_reader.hpp" +#include +#include +#include +#include +#include +#include +#include +#include + +namespace cpp_utils::io +{ + +// A range_reader over a local file that logs every call, to test and measure what a remote +// reader would be asked for. Copies share the file and the log. +class local_range_reader +{ + struct shared_state + { + explicit shared_state(const std::string& path) : file { path } { } + memory_mapped_file file; + mutable std::mutex mutex; + std::vector> batches; + std::vector threads; + std::size_t bytes_served = 0; + }; + std::shared_ptr state; + + void check_bounds(const byte_range& range) const + { + if (range.offset > size() || range.size > size() - range.offset) + throw std::out_of_range { "range past the end of the file at offset " + + std::to_string(range.offset) }; + } + + void log(std::span ranges) const + { + std::lock_guard lock { state->mutex }; + state->batches.emplace_back(std::begin(ranges), std::end(ranges)); + state->threads.push_back(std::this_thread::get_id()); + for (const auto& range : ranges) + state->bytes_served += range.size; + } + +public: + explicit local_range_reader(const std::string& path) + : state { std::make_shared(path) } + { + } + + [[nodiscard]] std::size_t size() const { return state->file.size(); } + + void read_ranges(std::span ranges, std::span destinations) const + { + for (const auto& range : ranges) + check_bounds(range); + log(ranges); + for (std::size_t i = 0; i < std::size(ranges); ++i) + std::memcpy(destinations[i], state->file.data() + ranges[i].offset, ranges[i].size); + } + + [[nodiscard]] std::vector> batches() const + { + std::lock_guard lock { state->mutex }; + return state->batches; + } + + [[nodiscard]] std::size_t bytes_served() const + { + std::lock_guard lock { state->mutex }; + return state->bytes_served; + } + + [[nodiscard]] std::vector caller_threads() const + { + std::lock_guard lock { state->mutex }; + return state->threads; + } +}; + +} diff --git a/include/cpp_utils/io/range_reader.hpp b/include/cpp_utils/io/range_reader.hpp new file mode 100644 index 0000000..8397194 --- /dev/null +++ b/include/cpp_utils/io/range_reader.hpp @@ -0,0 +1,50 @@ +/*------------------------------------------------------------------------------ +-- The MIT License (MIT) +-- +-- Copyright © 2024, Laboratory of Plasma Physics- CNRS +-- +-- Permission is hereby granted, free of charge, to any person obtaining a copy +-- of this software and associated documentation files (the “Software”), to deal +-- in the Software without restriction, including without limitation the rights +-- to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies +-- of the Software, and to permit persons to whom the Software is furnished to do +-- so, subject to the following conditions: +-- +-- The above copyright notice and this permission notice shall be included in all +-- copies or substantial portions of the Software. +-- +-- THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, +-- INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +-- PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +-- HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +-- OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE +-- SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +-------------------------------------------------------------------------------*/ +/*-- Author : Alexis Jeandet +-- Mail : alexis.jeandet@member.fsf.org +----------------------------------------------------------------------------*/ +#pragma once + +#include +#include +#include + +namespace cpp_utils::io +{ + +struct byte_range +{ + std::size_t offset; + std::size_t size; +}; + +// A source of bytes fetched by ranges (a remote object, a file...). read_ranges fills every +// destination with exactly its range's bytes, or throws: never a partial result. +template +concept range_reader = std::copy_constructible + && requires(const R& r, std::span ranges, std::span dest) { + { r.size() } -> std::convertible_to; + r.read_ranges(ranges, dest); + }; + +} diff --git a/tests/io/test_local_range_reader.cpp b/tests/io/test_local_range_reader.cpp new file mode 100644 index 0000000..6ec6f69 --- /dev/null +++ b/tests/io/test_local_range_reader.cpp @@ -0,0 +1,69 @@ +#include +#include +#include +#include +#include +#include +#include + +using namespace cpp_utils::io; + +static_assert(range_reader); + +namespace +{ +std::string write_temp_file(const std::string& name, const std::string& content) +{ + const auto path = (std::filesystem::temp_directory_path() / name).string(); + std::ofstream { path, std::ios::binary } << content; + return path; +} +} + +TEST_CASE("local_range_reader reads several ranges in one batch and logs it", "[io]") +{ + const auto path = write_temp_file("cpp_utils_local_range_reader.bin", "0123456789abcdef"); + local_range_reader reader { path }; + REQUIRE(reader.size() == 16); + + const byte_range ranges[] { { 0, 4 }, { 8, 4 } }; + char first[4] {}, second[4] {}; + char* const destinations[] { first, second }; + reader.read_ranges(ranges, destinations); + + REQUIRE(std::string(first, 4) == "0123"); + REQUIRE(std::string(second, 4) == "89ab"); + REQUIRE(reader.batches().size() == 1); + REQUIRE(reader.batches()[0].size() == 2); + REQUIRE(reader.bytes_served() == 8); + REQUIRE(reader.caller_threads().size() == 1); + std::filesystem::remove(path); +} + +TEST_CASE("copies of a local_range_reader share one call log", "[io]") +{ + const auto path = write_temp_file("cpp_utils_local_range_reader_copy.bin", "0123456789"); + local_range_reader reader { path }; + local_range_reader copy = reader; + + const byte_range ranges[] { { 2, 3 } }; + char dest[3] {}; + char* const destinations[] { dest }; + copy.read_ranges(ranges, destinations); + + REQUIRE(reader.batches().size() == 1); + REQUIRE(reader.bytes_served() == 3); + std::filesystem::remove(path); +} + +TEST_CASE("local_range_reader throws on a range past the end", "[io]") +{ + const auto path = write_temp_file("cpp_utils_local_range_reader_end.bin", "0123456789"); + local_range_reader reader { path }; + + const byte_range ranges[] { { 8, 4 } }; + char dest[4] {}; + char* const destinations[] { dest }; + REQUIRE_THROWS_AS(reader.read_ranges(ranges, destinations), std::out_of_range); + std::filesystem::remove(path); +} diff --git a/tests/meson.build b/tests/meson.build index 533c697..c7c445d 100644 --- a/tests/meson.build +++ b/tests/meson.build @@ -115,6 +115,11 @@ tests = [ 'sources': ['io/test_memory_mapped_file.cpp'], 'deps': [cpp_utils_dep, catch2_dep] }, + { + 'name':'TestLocalRangeReader', + 'sources': ['io/test_local_range_reader.cpp'], + 'deps': [cpp_utils_dep, catch2_dep] + }, { 'name':'TestBufferView', 'sources': ['io/test_buffer_view.cpp'], From cfb8fa2056145bdfb4b2582aa3af8fb1e54485f3 Mon Sep 17 00:00:00 2001 From: Alexis Jeandet Date: Mon, 2 Nov 2026 10:24:48 +0100 Subject: [PATCH 2/7] feat(io): reserved_region with page protection and memory/file backing Co-Authored-By: Claude Opus 5.5 --- include/cpp_utils/io/reserved_region.hpp | 349 +++++++++++++++++++++++ tests/io/test_reserved_region.cpp | 101 +++++++ tests/meson.build | 5 + 3 files changed, 455 insertions(+) create mode 100644 include/cpp_utils/io/reserved_region.hpp create mode 100644 tests/io/test_reserved_region.cpp diff --git a/include/cpp_utils/io/reserved_region.hpp b/include/cpp_utils/io/reserved_region.hpp new file mode 100644 index 0000000..f7e3d1b --- /dev/null +++ b/include/cpp_utils/io/reserved_region.hpp @@ -0,0 +1,349 @@ +/*------------------------------------------------------------------------------ +-- The MIT License (MIT) +-- +-- Copyright © 2024, Laboratory of Plasma Physics- CNRS +-- +-- Permission is hereby granted, free of charge, to any person obtaining a copy +-- of this software and associated documentation files (the “Software”), to deal +-- in the Software without restriction, including without limitation the rights +-- to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies +-- of the Software, and to permit persons to whom the Software is furnished to do +-- so, subject to the following conditions: +-- +-- The above copyright notice and this permission notice shall be included in all +-- copies or substantial portions of the Software. +-- +-- THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, +-- INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +-- PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +-- HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +-- OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE +-- SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +-------------------------------------------------------------------------------*/ +/*-- Author : Alexis Jeandet +-- Mail : alexis.jeandet@member.fsf.org +----------------------------------------------------------------------------*/ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +#if __has_include() +#define NOMINMAX +#include +#include +#define CPP_UTILS_REGION_WIN32 +#else +#include +#include +#include +#endif + +namespace cpp_utils::io +{ + +enum class region_backing +{ + memory, + file +}; + +namespace details::region +{ + [[noreturn]] inline void throw_last_error(const char* what) + { +#ifdef CPP_UTILS_REGION_WIN32 + throw std::system_error { static_cast(::GetLastError()), std::system_category(), what }; +#else + throw std::system_error { errno, std::generic_category(), what }; +#endif + } + + inline std::filesystem::path env_path(const char* name) + { + const char* value = std::getenv(name); + return value ? std::filesystem::path { value } : std::filesystem::path {}; + } + + inline std::filesystem::path user_cache_dir() + { +#if defined(CPP_UTILS_REGION_WIN32) + return env_path("LOCALAPPDATA"); +#elif defined(__APPLE__) + return env_path("HOME") / "Library" / "Caches"; +#else + if (auto xdg = env_path("XDG_CACHE_HOME"); !xdg.empty()) + return xdg; + return env_path("HOME") / ".cache"; +#endif + } + + inline std::filesystem::path resolve_temp_dir(const std::filesystem::path& requested) + { + auto dir = requested.empty() ? user_cache_dir() : requested; + if (dir.empty()) + throw std::runtime_error { "no directory for a file-backed region" }; + std::filesystem::create_directories(dir); + return dir; + } + +#ifdef CPP_UTILS_REGION_WIN32 + struct mapping + { + char* data = nullptr; + HANDLE file = INVALID_HANDLE_VALUE; + HANDLE section = nullptr; + }; + + inline mapping reserve_memory(std::size_t length) + { + auto* p = ::VirtualAlloc(nullptr, length, MEM_RESERVE, PAGE_NOACCESS); + if (!p) + throw_last_error("VirtualAlloc(MEM_RESERVE)"); + return { static_cast(p) }; + } + + inline HANDLE create_sparse_temp_file(const std::filesystem::path& dir, std::size_t length) + { + wchar_t name[MAX_PATH]; + if (!::GetTempFileNameW(dir.c_str(), L"cur", 0, name)) + throw_last_error("GetTempFileNameW"); + HANDLE file = ::CreateFileW(name, GENERIC_READ | GENERIC_WRITE, + FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, nullptr, CREATE_ALWAYS, + FILE_ATTRIBUTE_TEMPORARY | FILE_FLAG_DELETE_ON_CLOSE, nullptr); + if (file == INVALID_HANDLE_VALUE) + throw_last_error("CreateFileW"); + DWORD returned = 0; + ::DeviceIoControl(file, FSCTL_SET_SPARSE, nullptr, 0, nullptr, 0, &returned, nullptr); + LARGE_INTEGER size; + size.QuadPart = static_cast(length); + if (!::SetFilePointerEx(file, size, nullptr, FILE_BEGIN) || !::SetEndOfFile(file)) + { + ::CloseHandle(file); + throw_last_error("SetEndOfFile"); + } + return file; + } + + inline mapping reserve_file(std::size_t length, const std::filesystem::path& dir) + { + mapping m; + m.file = create_sparse_temp_file(dir, length); + m.section = ::CreateFileMappingW(m.file, nullptr, PAGE_READWRITE, 0, 0, nullptr); + if (m.section) + m.data = static_cast(::MapViewOfFile(m.section, FILE_MAP_WRITE, 0, 0, length)); + DWORD old = 0; + if (!m.data || !::VirtualProtect(m.data, length, PAGE_NOACCESS, &old)) + { + const auto error = ::GetLastError(); + if (m.data) + ::UnmapViewOfFile(m.data); + if (m.section) + ::CloseHandle(m.section); + ::CloseHandle(m.file); + ::SetLastError(error); + throw_last_error("MapViewOfFile"); + } + return m; + } + + inline void release(mapping& m, std::size_t) + { + if (m.section) + { + ::UnmapViewOfFile(m.data); + ::CloseHandle(m.section); + ::CloseHandle(m.file); + } + else + ::VirtualFree(m.data, 0, MEM_RELEASE); + } + + inline void protect(char* p, std::size_t length, DWORD protection) + { + DWORD old = 0; + if (!::VirtualProtect(p, length, protection, &old)) + throw_last_error("VirtualProtect"); + } + + inline void make_writable(const mapping& m, char* p, std::size_t length) + { + if (!m.section) + { + if (!::VirtualAlloc(p, length, MEM_COMMIT, PAGE_READWRITE)) + throw_last_error("VirtualAlloc(MEM_COMMIT)"); + return; + } + protect(p, length, PAGE_READWRITE); + } + + inline void seal(const mapping&, char* p, std::size_t length) + { + protect(p, length, PAGE_READONLY); + } + + inline void make_no_access(const mapping&, char* p, std::size_t length) + { + protect(p, length, PAGE_NOACCESS); + } + + inline std::size_t page_size() + { + SYSTEM_INFO info; + ::GetSystemInfo(&info); + return info.dwPageSize; + } +#else + struct mapping + { + char* data = nullptr; + }; + + inline mapping checked_mmap(void* p, const char* what) + { + if (p == MAP_FAILED) + throw_last_error(what); + return { static_cast(p) }; + } + + inline mapping reserve_memory(std::size_t length) + { + return checked_mmap(::mmap(nullptr, length, PROT_NONE, + MAP_PRIVATE | MAP_ANONYMOUS | MAP_NORESERVE, -1, 0), + "mmap(anonymous)"); + } + + // The file is unlinked at once: it disappears with the mapping, even after a crash. + inline int create_unlinked_temp_file(const std::filesystem::path& dir, std::size_t length) + { + auto name = (dir / "cpp_utils_region_XXXXXX").string(); + const int fd = ::mkstemp(name.data()); + if (fd == -1) + throw_last_error("mkstemp"); + ::unlink(name.c_str()); + if (::ftruncate(fd, static_cast(length)) != 0) + { + ::close(fd); + throw_last_error("ftruncate"); + } + return fd; + } + + inline mapping reserve_file(std::size_t length, const std::filesystem::path& dir) + { + const int fd = create_unlinked_temp_file(dir, length); + void* p = ::mmap(nullptr, length, PROT_NONE, MAP_SHARED, fd, 0); + ::close(fd); + return checked_mmap(p, "mmap(file)"); + } + + inline void release(mapping& m, std::size_t length) { ::munmap(m.data, length); } + + inline void protect(char* p, std::size_t length, int protection) + { + if (::mprotect(p, length, protection) != 0) + throw_last_error("mprotect"); + } + + inline void make_writable(const mapping&, char* p, std::size_t length) + { + protect(p, length, PROT_READ | PROT_WRITE); + } + + inline void seal(const mapping&, char* p, std::size_t length) { protect(p, length, PROT_READ); } + + inline void make_no_access(const mapping&, char* p, std::size_t length) + { + protect(p, length, PROT_NONE); + } + + inline std::size_t page_size() { return static_cast(::sysconf(_SC_PAGESIZE)); } +#endif +} + +[[nodiscard]] inline std::size_t page_size() +{ + static const std::size_t size = details::region::page_size(); + return size; +} + +// A region of address space the size of a file, whose pages are inaccessible until written and +// sealed. Reading a page that was never sealed crashes instead of returning zeros. +class reserved_region +{ + details::region::mapping p_mapping {}; + std::size_t p_size = 0; + std::size_t p_length = 0; + + [[nodiscard]] char* page(std::size_t index) const { return p_mapping.data + index * page_size(); } + [[nodiscard]] std::size_t pages_length(std::size_t first_page, std::size_t page_count) const + { + if (first_page + page_count > p_length / page_size()) + throw std::out_of_range { "page range outside the region" }; + return page_count * page_size(); + } + +public: + explicit reserved_region(std::size_t size, region_backing backing = region_backing::memory, + const std::filesystem::path& temp_dir = {}) + : p_size { size }, p_length { (size + page_size() - 1) / page_size() * page_size() } + { + if (p_length == 0) + return; + p_mapping = backing == region_backing::memory + ? details::region::reserve_memory(p_length) + : details::region::reserve_file(p_length, details::region::resolve_temp_dir(temp_dir)); + } + + reserved_region(const reserved_region&) = delete; + reserved_region& operator=(const reserved_region&) = delete; + + reserved_region(reserved_region&& other) noexcept + : p_mapping { std::exchange(other.p_mapping, {}) } + , p_size { std::exchange(other.p_size, 0) } + , p_length { std::exchange(other.p_length, 0) } + { + } + + reserved_region& operator=(reserved_region&& other) noexcept + { + std::swap(p_mapping, other.p_mapping); + std::swap(p_size, other.p_size); + std::swap(p_length, other.p_length); + return *this; + } + + ~reserved_region() + { + if (p_mapping.data) + details::region::release(p_mapping, p_length); + } + + [[nodiscard]] char* data() const { return p_mapping.data; } + [[nodiscard]] std::size_t size() const { return p_size; } + + char* make_writable(std::size_t first_page, std::size_t page_count) + { + details::region::make_writable( + p_mapping, page(first_page), pages_length(first_page, page_count)); + return page(first_page); + } + + void seal(std::size_t first_page, std::size_t page_count) + { + details::region::seal(p_mapping, page(first_page), pages_length(first_page, page_count)); + } + + void make_no_access(std::size_t first_page, std::size_t page_count) + { + details::region::make_no_access( + p_mapping, page(first_page), pages_length(first_page, page_count)); + } +}; + +} diff --git a/tests/io/test_reserved_region.cpp b/tests/io/test_reserved_region.cpp new file mode 100644 index 0000000..1e45c96 --- /dev/null +++ b/tests/io/test_reserved_region.cpp @@ -0,0 +1,101 @@ +#include +#include +#include +#include +#ifndef _WIN32 +#include +#include +#include +#endif + +using namespace cpp_utils::io; + +namespace +{ +constexpr region_backing all_backings[] { region_backing::memory, region_backing::file }; + +void write_page(reserved_region& region, std::size_t page, const std::string& text) +{ + std::memcpy(region.make_writable(page, 1), text.data(), std::size(text)); + region.seal(page, 1); +} + +#ifndef _WIN32 +// Reads one byte at offset in a child process; true when the read killed the child. +bool read_faults(const reserved_region& region, std::size_t offset) +{ + const pid_t pid = fork(); + if (pid == 0) + { + volatile char c = region.data()[offset]; + (void)c; + _exit(0); + } + int status = 0; + waitpid(pid, &status, 0); + return WIFSIGNALED(status) + && (WTERMSIG(status) == SIGSEGV || WTERMSIG(status) == SIGBUS); +} +#endif +} + +TEST_CASE("a sealed page reads back what was written", "[io]") +{ + for (auto backing : all_backings) + { + reserved_region region { 4 * page_size(), backing }; + write_page(region, 1, "hello"); + REQUIRE(std::string(region.data() + page_size(), 5) == "hello"); + } +} + +TEST_CASE("the last partial page of a region can be sealed", "[io]") +{ + for (auto backing : all_backings) + { + reserved_region region { 3 * page_size() + 7, backing }; + write_page(region, 3, "tail"); + REQUIRE(region.size() == 3 * page_size() + 7); + REQUIRE(std::string(region.data() + 3 * page_size(), 4) == "tail"); + } +} + +TEST_CASE("an empty region maps nothing", "[io]") +{ + for (auto backing : all_backings) + { + reserved_region region { 0, backing }; + REQUIRE(region.data() == nullptr); + REQUIRE(region.size() == 0); + } +} + +#ifndef _WIN32 +TEST_CASE("pages that were never sealed fault on read", "[io]") +{ + for (auto backing : all_backings) + { + reserved_region region { 4 * page_size(), backing }; + write_page(region, 1, "x"); + REQUIRE_FALSE(read_faults(region, page_size())); + REQUIRE(read_faults(region, 0)); + REQUIRE(read_faults(region, 2 * page_size())); + } +} + +TEST_CASE("make_no_access closes a sealed page again", "[io]") +{ + for (auto backing : all_backings) + { + reserved_region region { 2 * page_size(), backing }; + write_page(region, 0, "x"); + region.make_no_access(0, 1); + REQUIRE(read_faults(region, 0)); + } +} + +TEST_CASE("a file region in a directory that cannot be written throws", "[io]") +{ + REQUIRE_THROWS(reserved_region { page_size(), region_backing::file, "/proc" }); +} +#endif diff --git a/tests/meson.build b/tests/meson.build index c7c445d..98b746a 100644 --- a/tests/meson.build +++ b/tests/meson.build @@ -120,6 +120,11 @@ tests = [ 'sources': ['io/test_local_range_reader.cpp'], 'deps': [cpp_utils_dep, catch2_dep] }, + { + 'name':'TestReservedRegion', + 'sources': ['io/test_reserved_region.cpp'], + 'deps': [cpp_utils_dep, catch2_dep] + }, { 'name':'TestBufferView', 'sources': ['io/test_buffer_view.cpp'], From 8b92ec1ff3410e884006f4d589c324aeea2c6803 Mon Sep 17 00:00:00 2001 From: Alexis Jeandet Date: Mon, 2 Nov 2026 10:27:59 +0100 Subject: [PATCH 3/7] feat(io): remote_buffer page cache with read-ahead and batched fetches Co-Authored-By: Claude Opus 5.5 --- docs/api/io.md | 111 ++++++++ include/cpp_utils/io/remote_buffer.hpp | 368 +++++++++++++++++++++++++ tests/io/test_remote_buffer.cpp | 294 ++++++++++++++++++++ tests/io/test_reserved_region.cpp | 2 + tests/meson.build | 5 + 5 files changed, 780 insertions(+) create mode 100644 include/cpp_utils/io/remote_buffer.hpp create mode 100644 tests/io/test_remote_buffer.cpp diff --git a/docs/api/io.md b/docs/api/io.md index e6d90c7..82cd619 100644 --- a/docs/api/io.md +++ b/docs/api/io.md @@ -414,3 +414,114 @@ reader.read_ranges(ranges, destinations); reader.batches().size(); // 1 reader.bytes_served(); // 8 ``` + +## reserved_region + +```cpp +#include +``` + +A region of address space the size of a file, whose pages cannot be read until they are +written and sealed. Reading a page that was never sealed crashes, instead of returning zeros. + +```cpp +enum class region_backing { memory, file }; + +class reserved_region +{ +public: + explicit reserved_region(std::size_t size, region_backing backing = region_backing::memory, + const std::filesystem::path& temp_dir = {}); + char* data() const; + std::size_t size() const; + char* make_writable(std::size_t first_page, std::size_t page_count); + void seal(std::size_t first_page, std::size_t page_count); + void make_no_access(std::size_t first_page, std::size_t page_count); +}; + +std::size_t page_size(); +``` + +- `memory` backing reserves anonymous memory. `file` backing maps a temporary file that is + deleted at once, so the OS can write pages to disk and drop them from RAM. With an empty + `temp_dir`, the file goes in the user cache directory (`$XDG_CACHE_HOME` or `~/.cache`, + `~/Library/Caches`, `%LOCALAPPDATA%`). Pick a real disk: on tmpfs a file is just RAM. +- `make_writable` opens pages for writing and returns a pointer to the first one; `seal` makes + them read-only; `make_no_access` closes them again. A failed protection change throws + `std::system_error`. +- A size of 0 maps nothing: `data()` is `nullptr`. +- Moving is allowed; copying is not. + +```cpp +cpp_utils::io::reserved_region region { 3 * cpp_utils::io::page_size() + 7 }; +std::memcpy(region.make_writable(1, 1), "hello", 5); +region.seal(1, 1); +region.data()[cpp_utils::io::page_size()]; // 'h' +region.data()[0]; // crashes: page 0 was never sealed +``` + +## remote_buffer + +```cpp +#include +``` + +A lazily fetched, read-in-place copy of a `range_reader`'s bytes. It is meant for remote +files (S3...), where every request costs time and money. `ensure` fetches what is missing and +widens it to the read-ahead window, all in one `read_ranges` batch. Each byte is fetched at +most once. Bytes that were never fetched can't be read: `check_present` throws, and a raw read +through `data()` hits a no-access page. + +```cpp +struct remote_buffer_options +{ + std::size_t read_ahead = 256 * 1024; + std::size_t tail_prefetch = 256 * 1024; + region_backing backing = region_backing::memory; + std::filesystem::path temp_dir {}; + std::size_t download_whole_below = 0; + static remote_buffer_options from_env(std::string_view prefix); + static remote_buffer_options from_env(std::string_view prefix, remote_buffer_options defaults); +}; + +template +class remote_buffer +{ +public: + explicit remote_buffer(R reader, remote_buffer_options options = {}); + void ensure(std::span ranges) const; + void ensure(std::size_t offset, std::size_t size) const; + void check_present(std::size_t offset, std::size_t size) const; + void read(char* dest, std::size_t offset, std::size_t size) const; + const char* data() const; + const char* begin() const; + const char* end() const; + std::size_t size() const; + bool is_valid() const; + const R& reader() const; +}; +``` + +- `read_ahead`: each fetch is widened to aligned windows of this size, rounded up to whole + pages. +- `tail_prefetch`: the first `ensure` also fetches this many bytes from the end of the file, + in the same batch. Formats that write metadata last (CDF) need them early. +- `download_whole_below`: if the file is smaller than this, the first `ensure` fetches all of + it. 0 turns it off. +- `from_env("PREFIX_")` reads `PREFIX_READ_AHEAD`, `PREFIX_TAIL_PREFETCH`, `PREFIX_BACKING` + (`memory` or `file`), `PREFIX_TEMP_DIR` and `PREFIX_DOWNLOAD_WHOLE_BELOW`. A bad value + throws `std::invalid_argument`. +- A range outside the file throws `std::runtime_error`. That check can't overflow, even on + corrupt offsets. +- If the reader throws, `ensure` rethrows and the batch's pages stay unreadable. A later call + retries them. +- `ensure` is thread-safe. The reader is never called while the cache lock is held. Threads + that need pages another thread is fetching wait for that fetch, rather than fetching again. + +```cpp +cpp_utils::io::local_range_reader reader { "data.bin" }; +cpp_utils::io::remote_buffer buffer { reader }; +buffer.ensure(10, 4); // one batch: [0, 256 KiB) + the last 256 KiB +buffer.data()[12]; // read in place +buffer.ensure(20, 4); // already there: no fetch +``` diff --git a/include/cpp_utils/io/remote_buffer.hpp b/include/cpp_utils/io/remote_buffer.hpp new file mode 100644 index 0000000..b5ca7ac --- /dev/null +++ b/include/cpp_utils/io/remote_buffer.hpp @@ -0,0 +1,368 @@ +/*------------------------------------------------------------------------------ +-- The MIT License (MIT) +-- +-- Copyright © 2024, Laboratory of Plasma Physics- CNRS +-- +-- Permission is hereby granted, free of charge, to any person obtaining a copy +-- of this software and associated documentation files (the “Software”), to deal +-- in the Software without restriction, including without limitation the rights +-- to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies +-- of the Software, and to permit persons to whom the Software is furnished to do +-- so, subject to the following conditions: +-- +-- The above copyright notice and this permission notice shall be included in all +-- copies or substantial portions of the Software. +-- +-- THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, +-- INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +-- PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT +-- HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +-- OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE +-- SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +-------------------------------------------------------------------------------*/ +/*-- Author : Alexis Jeandet +-- Mail : alexis.jeandet@member.fsf.org +----------------------------------------------------------------------------*/ +#pragma once + +#include "range_reader.hpp" +#include "reserved_region.hpp" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace cpp_utils::io +{ + +struct remote_buffer_options +{ + std::size_t read_ahead = 256 * 1024; + std::size_t tail_prefetch = 256 * 1024; + region_backing backing = region_backing::memory; + std::filesystem::path temp_dir {}; + std::size_t download_whole_below = 0; + + // Reads READ_AHEAD, TAIL_PREFETCH, BACKING (memory|file), TEMP_DIR and + // DOWNLOAD_WHOLE_BELOW; unset or empty variables keep the given defaults. + [[nodiscard]] static remote_buffer_options from_env( + std::string_view prefix, remote_buffer_options defaults); + [[nodiscard]] static remote_buffer_options from_env(std::string_view prefix); + +private: + static std::string variable_name(std::string_view prefix, std::string_view name) + { + return std::string { prefix } + std::string { name }; + } + + static std::string env(std::string_view prefix, std::string_view name) + { + const char* value = std::getenv(variable_name(prefix, name).c_str()); + return value ? std::string { value } : std::string {}; + } + + static void read_size(std::string_view prefix, std::string_view name, std::size_t& target) + { + const auto value = env(prefix, name); + if (value.empty()) + return; + std::size_t parsed = 0; + const auto [end, error] = std::from_chars(value.data(), value.data() + value.size(), parsed); + if (error != std::errc {} || end != value.data() + value.size()) + throw std::invalid_argument { variable_name(prefix, name) + ": not a byte count: " + + value }; + target = parsed; + } + + static region_backing parse_backing(const std::string& name, const std::string& value) + { + if (value == "memory") + return region_backing::memory; + if (value == "file") + return region_backing::file; + throw std::invalid_argument { name + ": expected memory or file, got " + value }; + } +}; + +inline remote_buffer_options remote_buffer_options::from_env( + std::string_view prefix, remote_buffer_options defaults) +{ + read_size(prefix, "READ_AHEAD", defaults.read_ahead); + read_size(prefix, "TAIL_PREFETCH", defaults.tail_prefetch); + read_size(prefix, "DOWNLOAD_WHOLE_BELOW", defaults.download_whole_below); + if (const auto value = env(prefix, "TEMP_DIR"); !value.empty()) + defaults.temp_dir = value; + if (const auto value = env(prefix, "BACKING"); !value.empty()) + defaults.backing = parse_backing(variable_name(prefix, "BACKING"), value); + return defaults; +} + +inline remote_buffer_options remote_buffer_options::from_env(std::string_view prefix) +{ + return from_env(prefix, remote_buffer_options {}); +} + +// Lazily fetched copy of a range_reader's bytes, readable in place through data(). ensure() +// fetches what is missing, widened to the read-ahead window, in one batch; each byte is fetched +// at most once. Bytes never fetched are not readable: check_present() throws, and a raw read +// hits a no-access page. +template +class remote_buffer +{ + enum class page_state : std::uint8_t + { + absent, + in_flight, + present + }; + + struct page_run + { + std::size_t first; + std::size_t count; + }; + + struct fetch_plan + { + std::vector runs; + std::vector awaited; + }; + + R p_reader; + remote_buffer_options p_options; + std::size_t p_size; + mutable reserved_region p_region; + mutable std::mutex p_mutex; + mutable std::condition_variable p_page_changed; + mutable std::vector p_pages; + mutable bool p_first_fetch_planned = false; + + static std::size_t round_up(std::size_t value, std::size_t step) + { + return (value + step - 1) / step * step; + } + + static remote_buffer_options normalized(remote_buffer_options options) + { + options.read_ahead = std::max(page_size(), round_up(options.read_ahead, page_size())); + return options; + } + + void check_bounds(const byte_range& range) const + { + if (range.offset > p_size || range.size > p_size - range.offset) + throw std::runtime_error { "byte range outside the remote file at offset " + + std::to_string(range.offset) }; + } + + [[nodiscard]] std::pair widened_pages(const byte_range& range) const + { + const auto window = p_options.read_ahead; + const auto begin = range.offset / window * window; + const auto end = std::min(round_up(range.offset + range.size, window), p_size); + return { begin / page_size(), round_up(end, page_size()) / page_size() }; + } + + [[nodiscard]] std::pair exact_pages(const byte_range& range) const + { + return { range.offset / page_size(), + round_up(range.offset + range.size, page_size()) / page_size() }; + } + + [[nodiscard]] std::vector first_fetch_extras() const + { + if (p_size < p_options.download_whole_below) + return { { 0, p_size } }; + const auto tail = std::min(p_options.tail_prefetch, p_size); + if (tail == 0) + return {}; + return { { p_size - tail, tail } }; + } + + // Called with the lock held: claims the missing pages and lists the ones other calls are + // fetching. + [[nodiscard]] fetch_plan plan(std::span ranges) const + { + std::vector claimed(std::size(p_pages), false); + if (!std::exchange(p_first_fetch_planned, true)) + for (const auto& extra : first_fetch_extras()) + claim(exact_pages(extra), claimed); + fetch_plan result; + for (const auto& range : ranges) + { + if (range.size == 0) + continue; + claim(widened_pages(range), claimed); + const auto [first, last] = exact_pages(range); + for (auto page = first; page < last; ++page) + if (p_pages[page] == page_state::in_flight && !claimed[page]) + result.awaited.push_back(page); + } + result.runs = runs_of(claimed); + return result; + } + + void claim(std::pair pages, std::vector& claimed) const + { + for (auto page = pages.first; page < pages.second; ++page) + { + if (p_pages[page] == page_state::absent) + { + p_pages[page] = page_state::in_flight; + claimed[page] = true; + } + } + } + + static std::vector runs_of(const std::vector& claimed) + { + std::vector runs; + for (std::size_t page = 0; page < std::size(claimed); ++page) + { + if (!claimed[page]) + continue; + if (!runs.empty() && runs.back().first + runs.back().count == page) + ++runs.back().count; + else + runs.push_back({ page, 1 }); + } + return runs; + } + + [[nodiscard]] byte_range bytes_of(const page_run& run) const + { + const auto offset = run.first * page_size(); + return { offset, std::min(run.count * page_size(), p_size - offset) }; + } + + void mark(const std::vector& runs, page_state state) const + { + { + std::lock_guard lock { p_mutex }; + for (const auto& run : runs) + std::fill_n(std::begin(p_pages) + run.first, run.count, state); + } + p_page_changed.notify_all(); + } + + void close_pages(const std::vector& runs) const noexcept + { + for (const auto& run : runs) + { + try + { + p_region.make_no_access(run.first, run.count); + } + catch (...) + { + } + } + } + + void fetch(const std::vector& runs) const + { + std::vector ranges; + std::vector destinations; + try + { + for (const auto& run : runs) + { + ranges.push_back(bytes_of(run)); + destinations.push_back(p_region.make_writable(run.first, run.count)); + } + p_reader.read_ranges(ranges, destinations); + for (const auto& run : runs) + p_region.seal(run.first, run.count); + } + catch (...) + { + close_pages(runs); + mark(runs, page_state::absent); + throw; + } + mark(runs, page_state::present); + } + + // True when every awaited page arrived; false when one of the fetches failed. + [[nodiscard]] bool wait_for(const std::vector& pages) const + { + std::unique_lock lock { p_mutex }; + p_page_changed.wait(lock, + [&] + { + return std::ranges::none_of( + pages, [&](auto page) { return p_pages[page] == page_state::in_flight; }); + }); + return std::ranges::all_of( + pages, [&](auto page) { return p_pages[page] == page_state::present; }); + } + +public: + explicit remote_buffer(R reader, remote_buffer_options options = {}) + : p_reader { std::move(reader) } + , p_options { normalized(std::move(options)) } + , p_size { p_reader.size() } + , p_region { p_size, p_options.backing, p_options.temp_dir } + , p_pages(round_up(p_size, page_size()) / page_size(), page_state::absent) + { + } + + void ensure(std::span ranges) const + { + for (const auto& range : ranges) + check_bounds(range); + while (true) + { + fetch_plan next; + { + std::lock_guard lock { p_mutex }; + next = plan(ranges); + } + if (!next.runs.empty()) + fetch(next.runs); + if (wait_for(next.awaited)) + return; + } + } + + void ensure(std::size_t offset, std::size_t size) const + { + const byte_range range { offset, size }; + ensure(std::span { &range, 1 }); + } + + void check_present(std::size_t offset, std::size_t size) const + { + check_bounds({ offset, size }); + const auto [first, last] = exact_pages({ offset, size }); + std::lock_guard lock { p_mutex }; + for (auto page = first; page < last; ++page) + if (p_pages[page] != page_state::present) + throw std::runtime_error { "remote bytes not fetched at offset " + + std::to_string(std::max(offset, page * page_size())) }; + } + + void read(char* dest, std::size_t offset, std::size_t size) const + { + ensure(offset, size); + std::memcpy(dest, p_region.data() + offset, size); + } + + [[nodiscard]] const char* data() const { return p_region.data(); } + [[nodiscard]] const char* begin() const { return p_region.data(); } + [[nodiscard]] const char* end() const { return p_region.data() + p_size; } + [[nodiscard]] std::size_t size() const { return p_size; } + [[nodiscard]] bool is_valid() const { return p_size != 0; } + [[nodiscard]] const R& reader() const { return p_reader; } +}; + +} diff --git a/tests/io/test_remote_buffer.cpp b/tests/io/test_remote_buffer.cpp new file mode 100644 index 0000000..3855094 --- /dev/null +++ b/tests/io/test_remote_buffer.cpp @@ -0,0 +1,294 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#ifndef _WIN32 +#include +#include +#include +#endif + +using namespace cpp_utils::io; + +namespace +{ +constexpr std::size_t KiB = 1024; +constexpr std::size_t window = 256 * KiB; +constexpr std::size_t big_size = 5 * window + 123; + +char pattern(std::size_t i) { return static_cast((i * 7 + i / 251) & 0xFF); } + +std::string write_pattern_file(const std::string& name, std::size_t size) +{ + const auto path = (std::filesystem::temp_directory_path() / name).string(); + std::string content(size, '\0'); + for (std::size_t i = 0; i < size; ++i) + content[i] = pattern(i); + std::ofstream { path, std::ios::binary } << content; + return path; +} + +void set_env(const char* name, const char* value) +{ +#ifdef _WIN32 + _putenv_s(name, value); +#else + setenv(name, value, 1); +#endif +} + +bool bytes_match(const char* data, std::size_t offset, std::size_t size) +{ + for (std::size_t i = offset; i < offset + size; ++i) + if (data[i] != pattern(i)) + return false; + return true; +} + +// Throws on the n-th call until healed; copies share the switch. +struct flaky_reader +{ + local_range_reader inner; + std::shared_ptr> calls = std::make_shared>(0); + std::shared_ptr> healed = std::make_shared>(false); + int failing_call = 2; + + std::size_t size() const { return inner.size(); } + void read_ranges(std::span ranges, std::span dest) const + { + if (++*calls == failing_call && !*healed) + throw std::runtime_error { "transport failure" }; + inner.read_ranges(ranges, dest); + } +}; +static_assert(range_reader); + +#ifndef _WIN32 +bool read_faults(const char* p) +{ + const pid_t pid = fork(); + if (pid == 0) + { + std::signal(SIGSEGV, SIG_DFL); + std::signal(SIGBUS, SIG_DFL); + volatile char c = *p; + (void)c; + _exit(0); + } + int status = 0; + waitpid(pid, &status, 0); + return WIFSIGNALED(status) && (WTERMSIG(status) == SIGSEGV || WTERMSIG(status) == SIGBUS); +} +#endif + +struct temp_file +{ + std::string path; + temp_file(const std::string& name, std::size_t size) : path { write_pattern_file(name, size) } + { + } + ~temp_file() { std::filesystem::remove(path); } +}; +} + +TEST_CASE("the first ensure fetches the read-ahead window and the tail in one batch", "[io]") +{ + temp_file f { "cpp_utils_rb_first.bin", big_size }; + local_range_reader reader { f.path }; + remote_buffer buffer { reader }; + + buffer.ensure(10, 4); + + REQUIRE(bytes_match(buffer.data(), 10, 4)); + const auto batches = reader.batches(); + REQUIRE(batches.size() == 1); + REQUIRE(batches[0].size() == 2); + REQUIRE(batches[0][0].offset == 0); + REQUIRE(batches[0][0].size == window); + REQUIRE(batches[0][1].offset == (big_size - window) / page_size() * page_size()); + REQUIRE(batches[0][1].offset + batches[0][1].size == big_size); +} + +TEST_CASE("bytes already fetched are not fetched again", "[io]") +{ + temp_file f { "cpp_utils_rb_again.bin", big_size }; + local_range_reader reader { f.path }; + remote_buffer buffer { reader }; + + buffer.ensure(10, 4); + buffer.ensure(20, 4); + + REQUIRE(reader.batches().size() == 1); +} + +TEST_CASE("one ensure of far-apart ranges sends one batch with two runs", "[io]") +{ + temp_file f { "cpp_utils_rb_two_runs.bin", big_size }; + local_range_reader reader { f.path }; + remote_buffer buffer { reader, remote_buffer_options { .tail_prefetch = 0 } }; + + const byte_range ranges[] { { 10, 4 }, { 3 * window + 5, 4 } }; + buffer.ensure(ranges); + + REQUIRE(reader.batches().size() == 1); + REQUIRE(reader.batches()[0].size() == 2); + REQUIRE(bytes_match(buffer.data(), 3 * window + 5, 4)); +} + +TEST_CASE("check_present throws until the bytes are fetched", "[io]") +{ + temp_file f { "cpp_utils_rb_present.bin", big_size }; + local_range_reader reader { f.path }; + remote_buffer buffer { reader, remote_buffer_options { .tail_prefetch = 0 } }; + + REQUIRE_THROWS_AS(buffer.check_present(3 * window, 1), std::runtime_error); + buffer.ensure(3 * window, 1); + REQUIRE_NOTHROW(buffer.check_present(3 * window, 1)); +} + +TEST_CASE("ranges outside the file throw without overflowing", "[io]") +{ + temp_file f { "cpp_utils_rb_bounds.bin", big_size }; + local_range_reader reader { f.path }; + remote_buffer buffer { reader }; + + REQUIRE_THROWS_AS(buffer.ensure(big_size, 1), std::runtime_error); + REQUIRE_THROWS_AS(buffer.ensure(big_size - 1, 10), std::runtime_error); + constexpr auto huge = std::numeric_limits::max(); + REQUIRE_THROWS_AS(buffer.ensure(huge - 10, 20), std::runtime_error); + REQUIRE(reader.batches().empty()); +} + +TEST_CASE("a file smaller than the tail is fetched whole, without underflow", "[io]") +{ + temp_file f { "cpp_utils_rb_small.bin", 1000 }; + local_range_reader reader { f.path }; + remote_buffer buffer { reader }; + + buffer.ensure(0, 1); + + REQUIRE(reader.bytes_served() == 1000); + REQUIRE(bytes_match(buffer.data(), 0, 1000)); +} + +TEST_CASE("no byte is fetched twice over random ensures", "[io]") +{ + temp_file f { "cpp_utils_rb_random.bin", big_size }; + local_range_reader reader { f.path }; + remote_buffer buffer { reader, remote_buffer_options { .read_ahead = 4 * KiB } }; + + std::mt19937 rng { 42 }; + std::uniform_int_distribution offsets { 0, big_size - 1 }; + for (int i = 0; i < 200; ++i) + { + const auto offset = offsets(rng); + const auto size = std::min(1 + offsets(rng) % 20000, big_size - offset); + buffer.ensure(offset, size); + REQUIRE(bytes_match(buffer.data(), offset, size)); + } + REQUIRE(reader.bytes_served() <= big_size); +} + +TEST_CASE("a failed fetch leaves its pages unreadable and can be retried", "[io]") +{ + temp_file f { "cpp_utils_rb_flaky.bin", big_size }; + flaky_reader reader { local_range_reader { f.path } }; + remote_buffer buffer { reader, remote_buffer_options { .tail_prefetch = 0 } }; + + buffer.ensure(0, 1); + REQUIRE_THROWS_AS(buffer.ensure(3 * window, 1), std::runtime_error); + REQUIRE_THROWS_AS(buffer.check_present(3 * window, 1), std::runtime_error); +#ifndef _WIN32 + REQUIRE(read_faults(buffer.data() + 3 * window)); +#endif + *reader.healed = true; + buffer.ensure(3 * window, 1); + REQUIRE(bytes_match(buffer.data(), 3 * window, 1)); +} + +TEST_CASE("threads ensuring the same range share one fetch", "[io]") +{ + temp_file f { "cpp_utils_rb_threads.bin", big_size }; + local_range_reader reader { f.path }; + remote_buffer buffer { reader }; + + std::vector threads; + std::atomic matches = 0; + for (int i = 0; i < 8; ++i) + threads.emplace_back( + [&] + { + buffer.ensure(2 * window, 100); + matches += bytes_match(buffer.data(), 2 * window, 100); + }); + for (auto& t : threads) + t.join(); + + REQUIRE(matches == 8); + REQUIRE(reader.bytes_served() <= big_size); + REQUIRE(reader.batches().size() <= 2); +} + +TEST_CASE("download_whole_below fetches the whole file only when it is smaller", "[io]") +{ + temp_file f { "cpp_utils_rb_whole.bin", big_size }; + { + local_range_reader reader { f.path }; + remote_buffer buffer { reader, + remote_buffer_options { .tail_prefetch = 0, .download_whole_below = big_size + 1 } }; + buffer.ensure(0, 1); + REQUIRE(reader.bytes_served() == big_size); + } + { + local_range_reader reader { f.path }; + remote_buffer buffer { reader, + remote_buffer_options { .tail_prefetch = 0, .download_whole_below = big_size } }; + buffer.ensure(0, 1); + REQUIRE(reader.bytes_served() < big_size); + } +} + +TEST_CASE("a remote buffer over an empty source is invalid", "[io]") +{ + temp_file f { "cpp_utils_rb_empty.bin", 0 }; + local_range_reader reader { f.path }; + remote_buffer buffer { reader }; + REQUIRE_FALSE(buffer.is_valid()); +} + +TEST_CASE("read copies through ensure", "[io]") +{ + temp_file f { "cpp_utils_rb_read.bin", big_size }; + local_range_reader reader { f.path }; + const remote_buffer buffer { reader }; + char dest[16]; + buffer.read(dest, 4 * window, 16); + REQUIRE(bytes_match(buffer.data(), 4 * window, 16)); + for (std::size_t i = 0; i < 16; ++i) + REQUIRE(dest[i] == pattern(4 * window + i)); +} + +TEST_CASE("options can come from environment variables", "[io]") +{ + set_env("X_READ_AHEAD", "65536"); + set_env("X_BACKING", "file"); + const auto options = remote_buffer_options::from_env("X_"); + REQUIRE(options.read_ahead == 65536); + REQUIRE(options.backing == region_backing::file); + + set_env("X_READ_AHEAD", "abc"); + REQUIRE_THROWS_AS(remote_buffer_options::from_env("X_"), std::invalid_argument); + set_env("X_READ_AHEAD", ""); + set_env("X_BACKING", ""); +} diff --git a/tests/io/test_reserved_region.cpp b/tests/io/test_reserved_region.cpp index 1e45c96..cfade12 100644 --- a/tests/io/test_reserved_region.cpp +++ b/tests/io/test_reserved_region.cpp @@ -27,6 +27,8 @@ bool read_faults(const reserved_region& region, std::size_t offset) const pid_t pid = fork(); if (pid == 0) { + std::signal(SIGSEGV, SIG_DFL); + std::signal(SIGBUS, SIG_DFL); volatile char c = region.data()[offset]; (void)c; _exit(0); diff --git a/tests/meson.build b/tests/meson.build index 98b746a..c042dab 100644 --- a/tests/meson.build +++ b/tests/meson.build @@ -125,6 +125,11 @@ tests = [ 'sources': ['io/test_reserved_region.cpp'], 'deps': [cpp_utils_dep, catch2_dep] }, + { + 'name':'TestRemoteBuffer', + 'sources': ['io/test_remote_buffer.cpp'], + 'deps': [cpp_utils_dep, catch2_dep] + }, { 'name':'TestBufferView', 'sources': ['io/test_buffer_view.cpp'], From 6bf80dbfca98c204c401504c3e0c61cf4003dc73 Mon Sep 17 00:00:00 2001 From: Alexis Jeandet Date: Mon, 2 Nov 2026 10:36:12 +0100 Subject: [PATCH 4/7] fix(io): address review of remote_buffer and reserved_region Reject reader sizes and read_ahead values that overflow the page math, retry the first-fetch tail after a failed first fetch, make planning exception-safe, crash rather than leave a failed batch readable, fetch nothing for empty requests, and check page ranges without overflow. Tests are stricter: one batch for concurrent ensures, writes to sealed pages fault, file backing is really file-backed. Co-Authored-By: Claude Opus 5.5 --- docs/api/io.md | 16 ++++- include/cpp_utils/io/remote_buffer.hpp | 86 +++++++++++++----------- include/cpp_utils/io/reserved_region.hpp | 18 ++++- tests/io/test_remote_buffer.cpp | 85 ++++++++++++++++++++++- tests/io/test_reserved_region.cpp | 65 +++++++++++++++++- 5 files changed, 224 insertions(+), 46 deletions(-) diff --git a/docs/api/io.md b/docs/api/io.md index 82cd619..b49e7c8 100644 --- a/docs/api/io.md +++ b/docs/api/io.md @@ -2,11 +2,17 @@ Three byte-buffer types: `buffer_view`, a non-owning view over memory you already have; `memory_mapped_file`, a file-backed buffer opened via `mmap`/`MapViewOfFile`; and `owned_buffer`, -an owning buffer backed by `no_init_vector`. All satisfy the same `types::concepts::random_access_buffer` +an owning buffer backed by `no_init_vector`. These three satisfy the same `types::concepts::random_access_buffer` shape (`read`/`view`/`size`/`is_valid`) and `std::ranges::contiguous_range`, so they're interchangeable: any can be passed directly to `serde::deserialize`, and code written against one works unmodified against the others. +For remote sources there are also `range_reader` (a transport concept), `local_range_reader` (a +logging local transport for tests), `reserved_region` (page-protected memory) and `remote_buffer` +(a lazily fetched cache over a `range_reader`). `remote_buffer` is contiguous and readable in +place, but it has no `view()`, and it is neither copyable nor movable: hold it behind a +`std::shared_ptr`. + ## buffer_view ```cpp @@ -406,6 +412,7 @@ copy handed to another object still reports to the original. - A range past the end throws `std::out_of_range`. ```cpp +// data.bin: an existing file of at least 12 bytes cpp_utils::io::local_range_reader reader { "data.bin" }; const cpp_utils::io::byte_range ranges[] { { 0, 4 }, { 8, 4 } }; char a[4], b[4]; @@ -508,17 +515,22 @@ public: in the same batch. Formats that write metadata last (CDF) need them early. - `download_whole_below`: if the file is smaller than this, the first `ensure` fetches all of it. 0 turns it off. +- `backing` / `temp_dir`: where the fetched bytes live, see `reserved_region`. The default is + anonymous memory. - `from_env("PREFIX_")` reads `PREFIX_READ_AHEAD`, `PREFIX_TAIL_PREFETCH`, `PREFIX_BACKING` (`memory` or `file`), `PREFIX_TEMP_DIR` and `PREFIX_DOWNLOAD_WHOLE_BELOW`. A bad value throws `std::invalid_argument`. - A range outside the file throws `std::runtime_error`. That check can't overflow, even on corrupt offsets. - If the reader throws, `ensure` rethrows and the batch's pages stay unreadable. A later call - retries them. + retries them, including the tail or whole-file extras of a failed first fetch. +- A reader `size()` too large to map throws `std::length_error`, and so does a `read_ahead` + too large to round (`std::invalid_argument`). Zero-length requests fetch nothing. - `ensure` is thread-safe. The reader is never called while the cache lock is held. Threads that need pages another thread is fetching wait for that fetch, rather than fetching again. ```cpp +// data.bin: an existing file larger than 512 KiB cpp_utils::io::local_range_reader reader { "data.bin" }; cpp_utils::io::remote_buffer buffer { reader }; buffer.ensure(10, 4); // one batch: [0, 256 KiB) + the last 256 KiB diff --git a/include/cpp_utils/io/remote_buffer.hpp b/include/cpp_utils/io/remote_buffer.hpp index b5ca7ac..5443c02 100644 --- a/include/cpp_utils/io/remote_buffer.hpp +++ b/include/cpp_utils/io/remote_buffer.hpp @@ -34,6 +34,7 @@ #include #include #include +#include #include #include #include @@ -135,6 +136,7 @@ class remote_buffer { std::vector runs; std::vector awaited; + bool includes_first_extras = false; }; R p_reader; @@ -153,6 +155,8 @@ class remote_buffer static remote_buffer_options normalized(remote_buffer_options options) { + if (options.read_ahead > std::numeric_limits::max() / 2) + throw std::invalid_argument { "read_ahead is too large" }; options.read_ahead = std::max(page_size(), round_up(options.read_ahead, page_size())); return options; } @@ -188,39 +192,41 @@ class remote_buffer return { { p_size - tail, tail } }; } - // Called with the lock held: claims the missing pages and lists the ones other calls are - // fetching. + // Called with the lock held. Everything that can throw (allocations) happens before the + // first page changes state, so a failure here leaves no page claimed. [[nodiscard]] fetch_plan plan(std::span ranges) const { + fetch_plan result; std::vector claimed(std::size(p_pages), false); - if (!std::exchange(p_first_fetch_planned, true)) + result.includes_first_extras = !p_first_fetch_planned; + if (result.includes_first_extras) for (const auto& extra : first_fetch_extras()) - claim(exact_pages(extra), claimed); - fetch_plan result; + select_absent(exact_pages(extra), claimed); + for (const auto& range : ranges) + select_absent(widened_pages(range), claimed); for (const auto& range : ranges) { - if (range.size == 0) - continue; - claim(widened_pages(range), claimed); const auto [first, last] = exact_pages(range); for (auto page = first; page < last; ++page) if (p_pages[page] == page_state::in_flight && !claimed[page]) result.awaited.push_back(page); } result.runs = runs_of(claimed); + set_state(result.runs, page_state::in_flight); + p_first_fetch_planned = true; return result; } - void claim(std::pair pages, std::vector& claimed) const + void select_absent(std::pair pages, std::vector& claimed) const { for (auto page = pages.first; page < pages.second; ++page) - { - if (p_pages[page] == page_state::absent) - { - p_pages[page] = page_state::in_flight; - claimed[page] = true; - } - } + claimed[page] = claimed[page] || p_pages[page] == page_state::absent; + } + + void set_state(const std::vector& runs, page_state state) const noexcept + { + for (const auto& run : runs) + std::fill_n(std::begin(p_pages) + run.first, run.count, state); } static std::vector runs_of(const std::vector& claimed) @@ -244,52 +250,49 @@ class remote_buffer return { offset, std::min(run.count * page_size(), p_size - offset) }; } - void mark(const std::vector& runs, page_state state) const + void publish(const fetch_plan& plan, page_state state) const { { std::lock_guard lock { p_mutex }; - for (const auto& run : runs) - std::fill_n(std::begin(p_pages) + run.first, run.count, state); + set_state(plan.runs, state); + if (state == page_state::absent && plan.includes_first_extras) + p_first_fetch_planned = false; } p_page_changed.notify_all(); } - void close_pages(const std::vector& runs) const noexcept + // A page left readable after a failed fetch would serve partial bytes: if it cannot be + // closed again, crashing (noexcept) is the safe outcome. + void close_pages(std::span runs) const noexcept { for (const auto& run : runs) - { - try - { - p_region.make_no_access(run.first, run.count); - } - catch (...) - { - } - } + p_region.make_no_access(run.first, run.count); } - void fetch(const std::vector& runs) const + void fetch(const fetch_plan& plan) const { std::vector ranges; std::vector destinations; + std::size_t opened = 0; try { - for (const auto& run : runs) + for (const auto& run : plan.runs) { ranges.push_back(bytes_of(run)); destinations.push_back(p_region.make_writable(run.first, run.count)); + ++opened; } p_reader.read_ranges(ranges, destinations); - for (const auto& run : runs) + for (const auto& run : plan.runs) p_region.seal(run.first, run.count); } catch (...) { - close_pages(runs); - mark(runs, page_state::absent); + close_pages(std::span { plan.runs }.first(opened)); + publish(plan, page_state::absent); throw; } - mark(runs, page_state::present); + publish(plan, page_state::present); } // True when every awaited page arrived; false when one of the fetches failed. @@ -318,17 +321,22 @@ class remote_buffer void ensure(std::span ranges) const { + std::vector wanted; for (const auto& range : ranges) + { check_bounds(range); - while (true) + if (range.size != 0) + wanted.push_back(range); + } + while (!wanted.empty()) { fetch_plan next; { std::lock_guard lock { p_mutex }; - next = plan(ranges); + next = plan(wanted); } if (!next.runs.empty()) - fetch(next.runs); + fetch(next); if (wait_for(next.awaited)) return; } @@ -353,6 +361,8 @@ class remote_buffer void read(char* dest, std::size_t offset, std::size_t size) const { + if (size == 0) + return; ensure(offset, size); std::memcpy(dest, p_region.data() + offset, size); } diff --git a/include/cpp_utils/io/reserved_region.hpp b/include/cpp_utils/io/reserved_region.hpp index f7e3d1b..4fe5867 100644 --- a/include/cpp_utils/io/reserved_region.hpp +++ b/include/cpp_utils/io/reserved_region.hpp @@ -28,6 +28,7 @@ #include #include #include +#include #include #include #include @@ -117,7 +118,12 @@ namespace details::region FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, nullptr, CREATE_ALWAYS, FILE_ATTRIBUTE_TEMPORARY | FILE_FLAG_DELETE_ON_CLOSE, nullptr); if (file == INVALID_HANDLE_VALUE) + { + const auto error = ::GetLastError(); + ::DeleteFileW(name); + ::SetLastError(error); throw_last_error("CreateFileW"); + } DWORD returned = 0; ::DeviceIoControl(file, FSCTL_SET_SPARSE, nullptr, 0, nullptr, 0, &returned, nullptr); LARGE_INTEGER size; @@ -280,10 +286,18 @@ class reserved_region std::size_t p_size = 0; std::size_t p_length = 0; + static std::size_t whole_pages_length(std::size_t size) + { + if (size > std::numeric_limits::max() - page_size()) + throw std::length_error { "region size too large: " + std::to_string(size) }; + return (size + page_size() - 1) / page_size() * page_size(); + } + [[nodiscard]] char* page(std::size_t index) const { return p_mapping.data + index * page_size(); } [[nodiscard]] std::size_t pages_length(std::size_t first_page, std::size_t page_count) const { - if (first_page + page_count > p_length / page_size()) + const auto pages = p_length / page_size(); + if (page_count == 0 || first_page > pages || page_count > pages - first_page) throw std::out_of_range { "page range outside the region" }; return page_count * page_size(); } @@ -291,7 +305,7 @@ class reserved_region public: explicit reserved_region(std::size_t size, region_backing backing = region_backing::memory, const std::filesystem::path& temp_dir = {}) - : p_size { size }, p_length { (size + page_size() - 1) / page_size() * page_size() } + : p_size { size }, p_length { whole_pages_length(size) } { if (p_length == 0) return; diff --git a/tests/io/test_remote_buffer.cpp b/tests/io/test_remote_buffer.cpp index 3855094..439c168 100644 --- a/tests/io/test_remote_buffer.cpp +++ b/tests/io/test_remote_buffer.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -225,10 +226,12 @@ TEST_CASE("threads ensuring the same range share one fetch", "[io]") std::vector threads; std::atomic matches = 0; + std::latch start { 8 }; for (int i = 0; i < 8; ++i) threads.emplace_back( [&] { + start.arrive_and_wait(); buffer.ensure(2 * window, 100); matches += bytes_match(buffer.data(), 2 * window, 100); }); @@ -236,8 +239,8 @@ TEST_CASE("threads ensuring the same range share one fetch", "[io]") t.join(); REQUIRE(matches == 8); - REQUIRE(reader.bytes_served() <= big_size); - REQUIRE(reader.batches().size() <= 2); + REQUIRE(reader.batches().size() == 1); + REQUIRE(reader.caller_threads().size() == 1); } TEST_CASE("download_whole_below fetches the whole file only when it is smaller", "[io]") @@ -292,3 +295,81 @@ TEST_CASE("options can come from environment variables", "[io]") set_env("X_READ_AHEAD", ""); set_env("X_BACKING", ""); } + +namespace +{ +struct huge_reader +{ + std::size_t claimed_size; + std::size_t size() const { return claimed_size; } + void read_ranges(std::span, std::span) const + { + throw std::logic_error { "must not be called" }; + } +}; +} + +TEST_CASE("a reader claiming an impossible size is rejected", "[io]") +{ + constexpr auto huge = std::numeric_limits::max(); + REQUIRE_THROWS_AS(remote_buffer(huge_reader { huge }), std::length_error); + REQUIRE_THROWS_AS(remote_buffer(huge_reader { huge - 100 }), std::length_error); +} + +TEST_CASE("the tail is still fetched when the first fetch failed", "[io]") +{ + temp_file f { "cpp_utils_rb_first_fails.bin", big_size }; + flaky_reader reader { local_range_reader { f.path }, std::make_shared>(0), + std::make_shared>(false), 1 }; + remote_buffer buffer { reader }; + + REQUIRE_THROWS_AS(buffer.ensure(0, 1), std::runtime_error); + *reader.healed = true; + buffer.ensure(0, 1); + const auto batches = reader.inner.batches(); + REQUIRE(batches.size() == 1); + REQUIRE(batches[0].size() == 2); + REQUIRE(batches[0][1].offset + batches[0][1].size == big_size); +} + +TEST_CASE("an empty request fetches nothing", "[io]") +{ + temp_file f { "cpp_utils_rb_empty_request.bin", big_size }; + local_range_reader reader { f.path }; + remote_buffer buffer { reader }; + char dest[1]; + buffer.ensure(10, 0); + buffer.read(dest, 10, 0); + REQUIRE(reader.batches().empty()); +} + +TEST_CASE("read_ahead is rounded up to whole pages", "[io]") +{ + temp_file f { "cpp_utils_rb_rounding.bin", big_size }; + local_range_reader reader { f.path }; + remote_buffer buffer { reader, + remote_buffer_options { .read_ahead = page_size() + 1, .tail_prefetch = 0 } }; + buffer.ensure(0, 1); + REQUIRE(reader.bytes_served() == 2 * page_size()); +} + +TEST_CASE("an absurd read_ahead is rejected", "[io]") +{ + temp_file f { "cpp_utils_rb_absurd.bin", big_size }; + local_range_reader reader { f.path }; + REQUIRE_THROWS_AS(remote_buffer(reader, + remote_buffer_options { .read_ahead + = std::numeric_limits::max() }), + std::invalid_argument); +} + +TEST_CASE("a file-backed remote buffer reads back its bytes", "[io]") +{ + temp_file f { "cpp_utils_rb_file_backing.bin", big_size }; + local_range_reader reader { f.path }; + remote_buffer buffer { reader, + remote_buffer_options { + .backing = region_backing::file, .temp_dir = std::filesystem::temp_directory_path() } }; + buffer.ensure(3 * window, 100); + REQUIRE(bytes_match(buffer.data(), 3 * window, 100)); +} diff --git a/tests/io/test_reserved_region.cpp b/tests/io/test_reserved_region.cpp index cfade12..c250d64 100644 --- a/tests/io/test_reserved_region.cpp +++ b/tests/io/test_reserved_region.cpp @@ -1,6 +1,10 @@ #include #include +#include #include +#include +#include +#include #include #ifndef _WIN32 #include @@ -21,14 +25,16 @@ void write_page(reserved_region& region, std::size_t page, const std::string& te } #ifndef _WIN32 -// Reads one byte at offset in a child process; true when the read killed the child. -bool read_faults(const reserved_region& region, std::size_t offset) +// Touches one byte at offset in a child process; true when that killed the child. +bool access_faults(const reserved_region& region, std::size_t offset, bool write) { const pid_t pid = fork(); if (pid == 0) { std::signal(SIGSEGV, SIG_DFL); std::signal(SIGBUS, SIG_DFL); + if (write) + region.data()[offset] = 'w'; volatile char c = region.data()[offset]; (void)c; _exit(0); @@ -38,6 +44,29 @@ bool read_faults(const reserved_region& region, std::size_t offset) return WIFSIGNALED(status) && (WTERMSIG(status) == SIGSEGV || WTERMSIG(status) == SIGBUS); } + +bool read_faults(const reserved_region& region, std::size_t offset) +{ + return access_faults(region, offset, false); +} + +#ifdef __linux__ +// The /proc/self/maps line of the mapping holding p. +std::string mapping_of(const char* p) +{ + std::ifstream maps { "/proc/self/maps" }; + const auto address = reinterpret_cast(p); + for (std::string line; std::getline(maps, line);) + { + std::uintptr_t begin = 0, end = 0; + char dash = 0; + std::istringstream { line } >> std::hex >> begin >> dash >> end; + if (begin <= address && address < end) + return line; + } + return {}; +} +#endif #endif } @@ -101,3 +130,35 @@ TEST_CASE("a file region in a directory that cannot be written throws", "[io]") REQUIRE_THROWS(reserved_region { page_size(), region_backing::file, "/proc" }); } #endif + +#ifndef _WIN32 +TEST_CASE("a sealed page cannot be written", "[io]") +{ + for (auto backing : all_backings) + { + reserved_region region { 2 * page_size(), backing }; + write_page(region, 1, "x"); + REQUIRE(access_faults(region, page_size(), true)); + } +} +#endif + +#ifdef __linux__ +TEST_CASE("file backing maps a deleted temp file, memory backing does not", "[io]") +{ + reserved_region file_region { page_size(), region_backing::file, + std::filesystem::temp_directory_path() }; + reserved_region memory_region { page_size(), region_backing::memory }; + REQUIRE(mapping_of(file_region.data()).find("cpp_utils_region_") != std::string::npos); + REQUIRE(mapping_of(file_region.data()).find("(deleted)") != std::string::npos); + REQUIRE(mapping_of(memory_region.data()).find("cpp_utils_region_") == std::string::npos); +} +#endif + +TEST_CASE("page ranges outside the region throw without overflowing", "[io]") +{ + reserved_region region { 2 * page_size() }; + REQUIRE_THROWS_AS(region.seal(2, 1), std::out_of_range); + REQUIRE_THROWS_AS(region.seal(std::numeric_limits::max(), 2), std::out_of_range); + REQUIRE_THROWS_AS(region.seal(0, 0), std::out_of_range); +} From cf58c1b5a6b2f862d1106af5466b650239e6c2f8 Mon Sep 17 00:00:00 2001 From: Alexis Jeandet Date: Mon, 2 Nov 2026 11:41:53 +0100 Subject: [PATCH 5/7] feat(io): remote_buffer waits through the reader's blocking_wait A thread waiting for another thread's fetch can now release an outer lock (Python's GIL) that the fetching thread needs. Co-Authored-By: Claude Opus 5.5 --- docs/api/io.md | 4 +++ include/cpp_utils/io/remote_buffer.hpp | 26 ++++++++++++++--- tests/io/test_remote_buffer.cpp | 39 ++++++++++++++++++++++++++ 3 files changed, 65 insertions(+), 4 deletions(-) diff --git a/docs/api/io.md b/docs/api/io.md index b49e7c8..335ca9e 100644 --- a/docs/api/io.md +++ b/docs/api/io.md @@ -528,6 +528,10 @@ public: too large to round (`std::invalid_argument`). Zero-length requests fetch nothing. - `ensure` is thread-safe. The reader is never called while the cache lock is held. Threads that need pages another thread is fetching wait for that fetch, rather than fetching again. +- A reader may provide `template void blocking_wait(F&& wait) const`. That + waiting goes through it, so a reader whose callers hold an outer lock can release it while + `wait()` blocks. Example: a Python reader releases the GIL, which the fetching thread + needs. ```cpp // data.bin: an existing file larger than 512 KiB diff --git a/include/cpp_utils/io/remote_buffer.hpp b/include/cpp_utils/io/remote_buffer.hpp index 5443c02..14e751f 100644 --- a/include/cpp_utils/io/remote_buffer.hpp +++ b/include/cpp_utils/io/remote_buffer.hpp @@ -295,15 +295,33 @@ class remote_buffer publish(plan, page_state::present); } + // Readers whose callers hold an outer lock (Python's GIL) provide blocking_wait(f), to + // release it while f blocks: the fetching thread may need it. + template + void blocking_wait(F&& wait) const + { + if constexpr (requires { p_reader.blocking_wait(wait); }) + p_reader.blocking_wait(wait); + else + wait(); + } + // True when every awaited page arrived; false when one of the fetches failed. [[nodiscard]] bool wait_for(const std::vector& pages) const { - std::unique_lock lock { p_mutex }; - p_page_changed.wait(lock, + if (pages.empty()) + return true; + std::unique_lock lock { p_mutex, std::defer_lock }; + blocking_wait( [&] { - return std::ranges::none_of( - pages, [&](auto page) { return p_pages[page] == page_state::in_flight; }); + lock.lock(); + p_page_changed.wait(lock, + [&] + { + return std::ranges::none_of(pages, + [&](auto page) { return p_pages[page] == page_state::in_flight; }); + }); }); return std::ranges::all_of( pages, [&](auto page) { return p_pages[page] == page_state::present; }); diff --git a/tests/io/test_remote_buffer.cpp b/tests/io/test_remote_buffer.cpp index 439c168..9bc4fee 100644 --- a/tests/io/test_remote_buffer.cpp +++ b/tests/io/test_remote_buffer.cpp @@ -1,5 +1,6 @@ #include #include +#include #include #include #include @@ -373,3 +374,41 @@ TEST_CASE("a file-backed remote buffer reads back its bytes", "[io]") buffer.ensure(3 * window, 100); REQUIRE(bytes_match(buffer.data(), 3 * window, 100)); } + +namespace +{ +// A slow reader that wants to know when a thread blocks waiting for another one's fetch. +struct waiting_reader +{ + local_range_reader inner; + std::shared_ptr> waits = std::make_shared>(0); + + std::size_t size() const { return inner.size(); } + void read_ranges(std::span ranges, std::span dest) const + { + std::this_thread::sleep_for(std::chrono::milliseconds { 200 }); + inner.read_ranges(ranges, dest); + } + template + void blocking_wait(F&& wait) const + { + ++*waits; + wait(); + } +}; +} + +TEST_CASE("waiting for another thread's fetch goes through the reader's blocking_wait", "[io]") +{ + temp_file f { "cpp_utils_rb_blocking_wait.bin", big_size }; + waiting_reader reader { local_range_reader { f.path } }; + remote_buffer buffer { reader }; + + std::thread first { [&] { buffer.ensure(2 * window, 100); } }; + std::this_thread::sleep_for(std::chrono::milliseconds { 50 }); + buffer.ensure(2 * window, 100); + first.join(); + + REQUIRE(*reader.waits >= 1); + REQUIRE(reader.inner.batches().size() == 1); +} From 41cf7fb85218b31ed46b861f9bad9da02f081e11 Mon Sep 17 00:00:00 2001 From: Alexis Jeandet Date: Mon, 2 Nov 2026 11:55:20 +0100 Subject: [PATCH 6/7] test(io): blocking_wait test synchronised by latches, not sleeps; document the exactly-once contract Co-Authored-By: Claude Opus 5.5 --- docs/api/io.md | 2 +- tests/io/test_remote_buffer.cpp | 15 +++++++++++---- 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/docs/api/io.md b/docs/api/io.md index 335ca9e..320cdec 100644 --- a/docs/api/io.md +++ b/docs/api/io.md @@ -531,7 +531,7 @@ public: - A reader may provide `template void blocking_wait(F&& wait) const`. That waiting goes through it, so a reader whose callers hold an outer lock can release it while `wait()` blocks. Example: a Python reader releases the GIL, which the fetching thread - needs. + needs. It must call `wait()` exactly once, on the calling thread. ```cpp // data.bin: an existing file larger than 512 KiB diff --git a/tests/io/test_remote_buffer.cpp b/tests/io/test_remote_buffer.cpp index 9bc4fee..c513567 100644 --- a/tests/io/test_remote_buffer.cpp +++ b/tests/io/test_remote_buffer.cpp @@ -377,16 +377,19 @@ TEST_CASE("a file-backed remote buffer reads back its bytes", "[io]") namespace { -// A slow reader that wants to know when a thread blocks waiting for another one's fetch. +// A reader that holds its fetch until released, and counts the waits it is asked to run. struct waiting_reader { local_range_reader inner; std::shared_ptr> waits = std::make_shared>(0); + std::shared_ptr fetching = std::make_shared(1); + std::shared_ptr release = std::make_shared(1); std::size_t size() const { return inner.size(); } void read_ranges(std::span ranges, std::span dest) const { - std::this_thread::sleep_for(std::chrono::milliseconds { 200 }); + fetching->count_down(); + release->wait(); inner.read_ranges(ranges, dest); } template @@ -405,9 +408,13 @@ TEST_CASE("waiting for another thread's fetch goes through the reader's blocking remote_buffer buffer { reader }; std::thread first { [&] { buffer.ensure(2 * window, 100); } }; - std::this_thread::sleep_for(std::chrono::milliseconds { 50 }); - buffer.ensure(2 * window, 100); + reader.fetching->wait(); + std::thread second { [&] { buffer.ensure(2 * window, 100); } }; + while (*reader.waits == 0) + std::this_thread::yield(); + reader.release->count_down(); first.join(); + second.join(); REQUIRE(*reader.waits >= 1); REQUIRE(reader.inner.batches().size() == 1); From 29e44698ca9c0aec02c199e24f01cb6cd0b504ff Mon Sep 17 00:00:00 2001 From: Alexis Jeandet Date: Mon, 2 Nov 2026 12:18:12 +0100 Subject: [PATCH 7/7] =?UTF-8?q?fix(io):=20final=20review=20=E2=80=94=20kno?= =?UTF-8?q?wn-size=20constructor,=20empty=20range=20at=20EOF,=20destinatio?= =?UTF-8?q?n=20count?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit remote_buffer can take a size its caller already queried (a remote size() is a request), check_present accepts an empty range at the end of the file, and local_range_reader rejects mismatched destination counts. Docs: one exception per bullet. Co-Authored-By: Claude Opus 5.5 --- docs/api/io.md | 11 ++++-- include/cpp_utils/io/local_range_reader.hpp | 2 ++ include/cpp_utils/io/remote_buffer.hpp | 10 +++++- tests/io/test_local_range_reader.cpp | 12 +++++++ tests/io/test_remote_buffer.cpp | 37 +++++++++++++++++++++ 5 files changed, 68 insertions(+), 4 deletions(-) diff --git a/docs/api/io.md b/docs/api/io.md index 320cdec..2f0093f 100644 --- a/docs/api/io.md +++ b/docs/api/io.md @@ -409,7 +409,8 @@ copy handed to another object still reports to the original. - `batches()` — a snapshot of every `read_ranges` call, one vector of ranges per call. - `bytes_served()` — the total bytes handed out. - `caller_threads()` — the thread id of each call, in call order. -- A range past the end throws `std::out_of_range`. +- A range past the end throws `std::out_of_range`, and a destination count that differs + from the range count throws `std::invalid_argument`. ```cpp // data.bin: an existing file of at least 12 bytes @@ -496,6 +497,7 @@ class remote_buffer { public: explicit remote_buffer(R reader, remote_buffer_options options = {}); + remote_buffer(R reader, std::size_t size, remote_buffer_options options); void ensure(std::span ranges) const; void ensure(std::size_t offset, std::size_t size) const; void check_present(std::size_t offset, std::size_t size) const; @@ -524,8 +526,11 @@ public: corrupt offsets. - If the reader throws, `ensure` rethrows and the batch's pages stay unreadable. A later call retries them, including the tail or whole-file extras of a failed first fetch. -- A reader `size()` too large to map throws `std::length_error`, and so does a `read_ahead` - too large to round (`std::invalid_argument`). Zero-length requests fetch nothing. +- A `size()` too large to map throws `std::length_error`. +- A `read_ahead` too large to round up throws `std::invalid_argument`. +- Zero-length requests fetch nothing, and are always present. +- `remote_buffer(reader, size, options)` takes a size the caller already knows, so the + reader's `size()`, often a request, is not called again. - `ensure` is thread-safe. The reader is never called while the cache lock is held. Threads that need pages another thread is fetching wait for that fetch, rather than fetching again. - A reader may provide `template void blocking_wait(F&& wait) const`. That diff --git a/include/cpp_utils/io/local_range_reader.hpp b/include/cpp_utils/io/local_range_reader.hpp index 539e9da..f5b3b03 100644 --- a/include/cpp_utils/io/local_range_reader.hpp +++ b/include/cpp_utils/io/local_range_reader.hpp @@ -80,6 +80,8 @@ class local_range_reader void read_ranges(std::span ranges, std::span destinations) const { + if (std::size(destinations) != std::size(ranges)) + throw std::invalid_argument { "one destination per range expected" }; for (const auto& range : ranges) check_bounds(range); log(ranges); diff --git a/include/cpp_utils/io/remote_buffer.hpp b/include/cpp_utils/io/remote_buffer.hpp index 14e751f..afb410f 100644 --- a/include/cpp_utils/io/remote_buffer.hpp +++ b/include/cpp_utils/io/remote_buffer.hpp @@ -329,9 +329,15 @@ class remote_buffer public: explicit remote_buffer(R reader, remote_buffer_options options = {}) + : remote_buffer(reader, reader.size(), std::move(options)) + { + } + + // For callers that already know the size: a remote reader's size() is often a request. + remote_buffer(R reader, std::size_t size, remote_buffer_options options) : p_reader { std::move(reader) } , p_options { normalized(std::move(options)) } - , p_size { p_reader.size() } + , p_size { size } , p_region { p_size, p_options.backing, p_options.temp_dir } , p_pages(round_up(p_size, page_size()) / page_size(), page_state::absent) { @@ -369,6 +375,8 @@ class remote_buffer void check_present(std::size_t offset, std::size_t size) const { check_bounds({ offset, size }); + if (size == 0) + return; const auto [first, last] = exact_pages({ offset, size }); std::lock_guard lock { p_mutex }; for (auto page = first; page < last; ++page) diff --git a/tests/io/test_local_range_reader.cpp b/tests/io/test_local_range_reader.cpp index 6ec6f69..bff981c 100644 --- a/tests/io/test_local_range_reader.cpp +++ b/tests/io/test_local_range_reader.cpp @@ -67,3 +67,15 @@ TEST_CASE("local_range_reader throws on a range past the end", "[io]") REQUIRE_THROWS_AS(reader.read_ranges(ranges, destinations), std::out_of_range); std::filesystem::remove(path); } + +TEST_CASE("local_range_reader rejects fewer destinations than ranges", "[io]") +{ + const auto path = write_temp_file("cpp_utils_local_range_reader_dest.bin", "0123456789"); + local_range_reader reader { path }; + const byte_range ranges[] { { 0, 2 }, { 4, 2 } }; + char dest[2] {}; + char* const destinations[] { dest }; + REQUIRE_THROWS_AS(reader.read_ranges(ranges, destinations), std::invalid_argument); + REQUIRE(reader.batches().empty()); + std::filesystem::remove(path); +} diff --git a/tests/io/test_remote_buffer.cpp b/tests/io/test_remote_buffer.cpp index c513567..cdc82d6 100644 --- a/tests/io/test_remote_buffer.cpp +++ b/tests/io/test_remote_buffer.cpp @@ -419,3 +419,40 @@ TEST_CASE("waiting for another thread's fetch goes through the reader's blocking REQUIRE(*reader.waits >= 1); REQUIRE(reader.inner.batches().size() == 1); } + +TEST_CASE("an empty range at the end of the file is always present", "[io]") +{ + temp_file f { "cpp_utils_rb_empty_at_end.bin", big_size }; + local_range_reader reader { f.path }; + remote_buffer buffer { reader }; + REQUIRE_NOTHROW(buffer.check_present(big_size, 0)); +} + +namespace +{ +struct size_counting_reader +{ + local_range_reader inner; + std::shared_ptr> size_calls = std::make_shared>(0); + + std::size_t size() const + { + ++*size_calls; + return inner.size(); + } + void read_ranges(std::span ranges, std::span dest) const + { + inner.read_ranges(ranges, dest); + } +}; +} + +TEST_CASE("a known size spares the reader its size query", "[io]") +{ + temp_file f { "cpp_utils_rb_known_size.bin", big_size }; + size_counting_reader reader { local_range_reader { f.path } }; + remote_buffer buffer { reader, big_size, remote_buffer_options {} }; + buffer.ensure(0, 1); + REQUIRE(*reader.size_calls == 0); + REQUIRE(bytes_match(buffer.data(), 0, 1)); +}