Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
189 changes: 188 additions & 1 deletion docs/api/io.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -359,3 +365,184 @@ writer.offset(); // 6

// file closed automatically when writer is destroyed
```

## range_reader

```cpp
#include <io/range_reader.hpp>
```

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 <typename R>
concept range_reader = std::copy_constructible<R> && requires(const R& r,
std::span<const byte_range> ranges, std::span<char* const> destinations) {
{ r.size() } -> std::convertible_to<std::size_t>;
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 <io/local_range_reader.hpp>
```

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`, 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
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
```

## reserved_region

```cpp
#include <io/reserved_region.hpp>
```

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 <io/remote_buffer.hpp>
```

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 <range_reader R>
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<const byte_range> 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.
- `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, including the tail or whole-file extras of a failed first fetch.
- 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 <typename F> 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. It must call `wait()` exactly once, on the calling thread.

```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
buffer.data()[12]; // read in place
buffer.ensure(20, 4); // already there: no fetch
```
111 changes: 111 additions & 0 deletions include/cpp_utils/io/local_range_reader.hpp
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
/*------------------------------------------------------------------------------
-- 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 <cstring>
#include <memory>
#include <mutex>
#include <span>
#include <stdexcept>
#include <string>
#include <thread>
#include <vector>

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<std::vector<byte_range>> batches;
std::vector<std::thread::id> threads;
std::size_t bytes_served = 0;
};
std::shared_ptr<shared_state> 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<const byte_range> 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<shared_state>(path) }
{
}

[[nodiscard]] std::size_t size() const { return state->file.size(); }

void read_ranges(std::span<const byte_range> ranges, std::span<char* const> 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);
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<std::vector<byte_range>> 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<std::thread::id> caller_threads() const
{
std::lock_guard lock { state->mutex };
return state->threads;
}
};

}
50 changes: 50 additions & 0 deletions include/cpp_utils/io/range_reader.hpp
Original file line number Diff line number Diff line change
@@ -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 <concepts>
#include <cstddef>
#include <span>

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 <typename R>
concept range_reader = std::copy_constructible<R>
&& requires(const R& r, std::span<const byte_range> ranges, std::span<char* const> dest) {
{ r.size() } -> std::convertible_to<std::size_t>;
r.read_ranges(ranges, dest);
};

}
Loading
Loading