sami/vdm
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

core: io/sparse_file + io/write_buffer (stage 4)

Browse Source 
io/sparse_file — the single O_WRONLY output file (docs/04 §4). open()
posix_fallocate's the full size (falls back to ftruncate on
EOPNOTSUPP/ENOSYS, reported via preallocated()); write_at() pwrites at an
absolute offset, looping short writes and retrying EINTR; sync() is
fdatasync (timer/pause only); advise_dontneed() is
posix_fadvise(DONTNEED); resize() trims a preallocated tail or sizes a
chunked download. errno -> vdm::Error (ENOSPC->disk_full,
EACCES->permission_denied, ENOENT/ENOTDIR/...->path_rejected). No lock on
write_at — POSIX makes each pwrite atomic for a regular file, so N
segment threads writing disjoint ranges is safe (tested, TSan-clean).

io/write_buffer — per-segment accumulate-and-flush buffer, preallocated
at construction; append() only memcpys (no allocation on the write-
callback hot path, docs/04 §8 — asserted by a global-new counter in the
test). Flushes on fill via a caller-supplied FlushFn; a chunk >= capacity
arriving on an empty buffer writes straight through. On a flush error
next_offset() stays at the last durable position. Single-threaded; the
disk-writer-thread handoff is stage 8.

Also: vtest.hpp VT_CHECK_EQ/NE now copy operands (auto, not auto&&) — an
assertion must not outlive a temporary the expression returned a
reference into (ASan caught this on Result<void>{}.error().code).

Co-Authored-By: Claude Sonnet 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01HPPSGhiArbvQgwC2DNiURS
This commit is contained in:
sami
2026-09-10 00:41:31 +04:00
co-authored by Claude Sonnet 5
parent e8a6b64c2f
commit bd0a24c87a
9 changed files with 781 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/tests/io/sparse_file_test.cpp
+181
View File
@@ -0,0 +1,181 @@
#include "vdm/io/sparse_file.hpp"
#include <fcntl.h>
#include <unistd.h>
#include <array>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <string>
#include <thread>
#include <vector>
#include "vtest.hpp"
using vdm::Error;
using vdm::io::SparseFile;
namespace {
struct TempPath {
std::string path;
TempPath() {
const char *dir = std::getenv("TMPDIR");
path = (dir ? dir : "/tmp");
path += "/vdm_sparse_test_XXXXXX";
int fd = ::mkstemp(path.data());
if (fd >= 0) {
::close(fd);
::unlink(path.c_str()); // we only wanted a unique name
}
}
~TempPath() { ::unlink(path.c_str()); }
};
std::vector<std::byte> read_all(const std::string &path) {
int fd = ::open(path.c_str(), O_RDONLY);
if (fd < 0)
return {};
std::vector<std::byte> out;
std::array<std::byte, 4096> buf{};
for (;;) {
ssize_t n = ::read(fd, buf.data(), buf.size());
if (n <= 0)
break;
out.insert(out.end(), buf.begin(), buf.begin() + n);
}
::close(fd);
return out;
}
std::uint64_t file_size(const std::string &path) {
int fd = ::open(path.c_str(), O_RDONLY);
if (fd < 0)
return 0;
off_t end = ::lseek(fd, 0, SEEK_END);
::close(fd);
return end < 0 ? 0 : static_cast<std::uint64_t>(end);
}
vdm::ConstByteSpan bytes(const char *s) {
return {reinterpret_cast<const std::byte *>(s), std::strlen(s)};
}
} // namespace
VT_TEST(sparse_open_preallocates_full_size) {
TempPath tp;
SparseFile f;
auto r = f.open(tp.path, {.total_size = 1 << 20});
VT_REQUIRE(r.has_value());
VT_CHECK(f.is_open());
VT_CHECK_EQ(file_size(tp.path), 1u << 20);
// /tmp is usually a real fs; if it's tmpfs, preallocated() is false and that's fine.
VT_CHECK(f.close().has_value());
}
VT_TEST(sparse_write_at_absolute_offsets) {
TempPath tp;
SparseFile f;
VT_REQUIRE(f.open(tp.path, {.total_size = 64}).has_value());
VT_CHECK(f.write_at(10, bytes("hello")).has_value());
VT_CHECK(f.write_at(40, bytes("world")).has_value());
VT_CHECK(f.sync().has_value());
auto data = read_all(tp.path);
VT_REQUIRE(data.size() == 64);
VT_CHECK_EQ(std::memcmp(data.data() + 10, "hello", 5), 0);
VT_CHECK_EQ(std::memcmp(data.data() + 40, "world", 5), 0);
f.close().value();
}
VT_TEST(sparse_write_past_end_grows_file) {
TempPath tp;
SparseFile f;
VT_REQUIRE(f.open(tp.path, {.total_size = 16}).has_value());
VT_CHECK(f.write_at(1000, bytes("tail")).has_value());
VT_CHECK_EQ(file_size(tp.path), 1004u);
f.close().value();
}
VT_TEST(sparse_resize_trims) {
TempPath tp;
SparseFile f;
VT_REQUIRE(f.open(tp.path, {.total_size = 4096}).has_value());
VT_CHECK(f.resize(100).has_value());
VT_CHECK_EQ(file_size(tp.path), 100u);
f.close().value();
}
VT_TEST(sparse_open_bad_path_is_path_rejected) {
SparseFile f;
auto r = f.open("/vdm_no_such_dir_xyz/file.part", {.total_size = 10});
VT_REQUIRE(!r.has_value());
VT_CHECK_EQ(r.error().code, Error::path_rejected);
VT_CHECK(!f.is_open());
}
VT_TEST(sparse_ops_on_closed_file_error) {
SparseFile f;
VT_CHECK_EQ(f.write_at(0, bytes("x")).error().code, Error::internal);
VT_CHECK_EQ(f.sync().error().code, Error::internal);
VT_CHECK(f.close().has_value()); // close on a closed file is ok
}
VT_TEST(sparse_advise_dontneed_is_safe) {
TempPath tp;
SparseFile f;
VT_REQUIRE(f.open(tp.path, {.total_size = 8192}).has_value());
VT_CHECK(f.write_at(0, bytes("data")).has_value());
VT_CHECK(f.sync().has_value());
f.advise_dontneed(0, 4096); // must not crash / must be a no-op-safe call
f.advise_dontneed(0, 0);
f.close().value();
}
VT_TEST(sparse_move_transfers_fd) {
TempPath tp;
SparseFile a;
VT_REQUIRE(a.open(tp.path, {.total_size = 32}).has_value());
SparseFile b = std::move(a);
VT_CHECK(!a.is_open());
VT_CHECK(b.is_open());
VT_CHECK(b.write_at(0, bytes("moved")).has_value());
b.close().value();
}
VT_TEST(sparse_concurrent_nonoverlapping_writes) {
TempPath tp;
SparseFile f;
constexpr int kSegs = 8;
constexpr std::size_t kSeg = 64 * 1024;
VT_REQUIRE(f.open(tp.path, {.total_size = kSegs * kSeg}).has_value());
std::vector<std::jthread> ts;
for (int s = 0; s < kSegs; ++s) {
ts.emplace_back([&, s] {
std::vector<std::byte> chunk(kSeg, static_cast<std::byte>('A' + s));
for (std::size_t off = 0; off < kSeg; off += 4096) {
auto r = f.write_at(static_cast<std::uint64_t>(s) * kSeg + off,
vdm::ConstByteSpan(chunk.data() + off, 4096));
if (!r.has_value())
VT_FAIL("concurrent write_at failed");
}
});
}
ts.clear(); // join
VT_CHECK(f.sync().has_value());
auto data = read_all(tp.path);
VT_REQUIRE(data.size() == kSegs * kSeg);
for (int s = 0; s < kSegs; ++s) {
bool ok = true;
for (std::size_t i = 0; i < kSeg; ++i)
if (data[s * kSeg + i] != static_cast<std::byte>('A' + s))
ok = false;
VT_CHECK(ok);
}
f.close().value();
}
core/tests/io/write_buffer_test.cpp
+202
View File
@@ -0,0 +1,202 @@
#include "vdm/io/write_buffer.hpp"
#include <atomic>
#include <cstdlib>
#include <cstring>
#include <new>
#include <string>
#include <vector>
#include "vtest.hpp"
using vdm::ConstByteSpan;
using vdm::Error;
using vdm::Result;
using vdm::io::WriteBuffer;
// --- global allocation counter, for the "no alloc in append()" test -----------------
namespace {
std::atomic<long> g_alloc_calls{0};
std::atomic<bool> g_count_allocs{false};
} // namespace
void *operator new(std::size_t n) {
if (g_count_allocs.load(std::memory_order_relaxed))
g_alloc_calls.fetch_add(1, std::memory_order_relaxed);
void *p = std::malloc(n ? n : 1);
if (!p)
throw std::bad_alloc();
return p;
}
void operator delete(void *p) noexcept {
std::free(p);
}
void operator delete(void *p, std::size_t) noexcept {
std::free(p);
}
void *operator new[](std::size_t n) {
return ::operator new(n);
}
void operator delete[](void *p) noexcept {
std::free(p);
}
void operator delete[](void *p, std::size_t) noexcept {
std::free(p);
}
namespace {
// A flush sink that records (offset, bytes) and never allocates after construction.
struct Sink {
std::vector<std::byte> data; // pre-reserved
std::vector<std::uint64_t> offs; // pre-reserved
std::vector<std::size_t> lens;
bool fail_next = false;
WriteBuffer::FlushFn fn() {
return [this](std::uint64_t off, ConstByteSpan s) -> Result<void> {
if (fail_next) {
fail_next = false;
return vdm::Err{Error::io_error, "sink forced failure"};
}
offs.push_back(off);
lens.push_back(s.size());
data.insert(data.end(), s.begin(), s.end());
return vdm::ok();
};
}
};
ConstByteSpan sv(const char *s) {
return {reinterpret_cast<const std::byte *>(s), std::strlen(s)};
}
} // namespace
VT_TEST(wb_accumulates_then_flushes_on_fill) {
Sink sink;
sink.data.reserve(1 << 16);
sink.offs.reserve(64);
sink.lens.reserve(64);
WriteBuffer wb(0, 8, sink.fn());
VT_CHECK(wb.append(sv("abc")).has_value()); // 3 buffered
VT_CHECK_EQ(wb.pending(), 3u);
VT_CHECK(sink.offs.empty()); // no flush yet
VT_CHECK(wb.append(sv("defgh")).has_value()); // fills to 8 -> flush
VT_REQUIRE(sink.offs.size() == 1);
VT_CHECK_EQ(sink.offs[0], 0u);
VT_CHECK_EQ(sink.lens[0], 8u);
VT_CHECK_EQ(wb.pending(), 0u);
VT_CHECK_EQ(wb.next_offset(), 8u);
VT_CHECK(wb.append(sv("ij")).has_value());
VT_CHECK(wb.flush().has_value()); // explicit tail flush
VT_REQUIRE(sink.offs.size() == 2);
VT_CHECK_EQ(sink.offs[1], 8u);
VT_CHECK_EQ(sink.lens[1], 2u);
VT_CHECK_EQ(std::string(reinterpret_cast<const char *>(sink.data.data()), sink.data.size()),
std::string("abcdefghij"));
VT_CHECK_EQ(wb.total_appended(), 10u);
}
VT_TEST(wb_flush_is_noop_when_empty) {
Sink sink;
sink.offs.reserve(4);
WriteBuffer wb(100, 16, sink.fn());
VT_CHECK(wb.flush().has_value());
VT_CHECK(sink.offs.empty());
}
VT_TEST(wb_oversized_chunk_writes_through) {
Sink sink;
sink.data.reserve(1 << 16);
sink.offs.reserve(16);
sink.lens.reserve(16);
WriteBuffer wb(0, 8, sink.fn());
VT_CHECK(wb.append(sv("ab")).has_value()); // 2 buffered
// 20 bytes arriving: buffer isn't empty, so first 6 top it off + flush(8), then the
// remaining 14 (>= capacity, buffer now empty) write straight through.
std::string big(20, 'x');
VT_CHECK(wb.append(sv(big.c_str())).has_value());
VT_CHECK(wb.flush().has_value());
// reconstruct
std::string got(reinterpret_cast<const char *>(sink.data.data()), sink.data.size());
VT_CHECK_EQ(got, std::string("ab") + big);
VT_CHECK_EQ(wb.total_appended(), 22u);
// one full-buffer flush + one passthrough; order preserved
VT_CHECK(sink.offs.size() >= 2);
VT_CHECK_EQ(sink.offs.front(), 0u);
}
VT_TEST(wb_exact_capacity_chunk_from_empty_writes_through) {
Sink sink;
sink.data.reserve(64);
sink.offs.reserve(4);
sink.lens.reserve(4);
WriteBuffer wb(0, 4, sink.fn());
VT_CHECK(wb.append(sv("wxyz")).has_value()); // == capacity, empty -> passthrough
VT_REQUIRE(sink.offs.size() == 1);
VT_CHECK_EQ(sink.lens[0], 4u);
VT_CHECK_EQ(wb.pending(), 0u);
}
VT_TEST(wb_flush_error_propagates_without_advancing_durable_offset) {
Sink sink;
sink.data.reserve(64);
sink.offs.reserve(4);
sink.lens.reserve(4);
WriteBuffer wb(0, 8, sink.fn());
VT_CHECK(wb.append(sv("abc")).has_value()); // 3 buffered, nothing durable yet
VT_CHECK_EQ(wb.next_offset(), 0u);
sink.fail_next = true;
auto r = wb.flush(); // forced failure
VT_REQUIRE(!r.has_value());
VT_CHECK_EQ(r.error().code, Error::io_error);
VT_CHECK_EQ(wb.next_offset(), 0u); // durable offset did NOT move
VT_CHECK(sink.offs.empty());
// a retry flush succeeds and advances
VT_CHECK(wb.flush().has_value());
VT_CHECK_EQ(wb.next_offset(), 3u);
VT_REQUIRE(sink.lens.size() == 1);
VT_CHECK_EQ(sink.lens[0], 3u);
}
VT_TEST(wb_append_does_not_allocate) {
// flush sink that never allocates: just sum sizes.
std::atomic<std::uint64_t> total{0};
auto flush = [&total](std::uint64_t, ConstByteSpan s) -> Result<void> {
total.fetch_add(s.size());
return vdm::ok();
};
WriteBuffer wb(0, 4096, flush);
// Warm up (any first-call lazy init happens now, outside the measured window).
std::string warm(100, 'w');
(void)wb.append(sv(warm.c_str()));
(void)wb.flush();
std::vector<std::byte> chunk(512, std::byte{7});
g_alloc_calls.store(0);
g_count_allocs.store(true);
for (int i = 0; i < 5000; ++i) {
auto r = wb.append(ConstByteSpan(chunk.data(), 137 + (i % 200)));
if (!r.has_value()) {
g_count_allocs.store(false);
VT_FAIL("append failed");
return;
}
}
(void)wb.flush();
g_count_allocs.store(false);
VT_CHECK_EQ(g_alloc_calls.load(), 0L);
VT_CHECK(total.load() > 0);
}