diff --git a/core/CMakeLists.txt b/core/CMakeLists.txt index 16dd02c..5cb5b82 100644 --- a/core/CMakeLists.txt +++ b/core/CMakeLists.txt @@ -17,6 +17,8 @@ add_library(veloxcore STATIC src/net/content_disposition.cpp src/net/url.cpp src/net/probe.cpp + src/io/sparse_file.cpp + src/io/write_buffer.cpp ) add_library(velox::core ALIAS veloxcore) diff --git a/core/include/vdm/io/sparse_file.hpp b/core/include/vdm/io/sparse_file.hpp new file mode 100644 index 0000000..1b7db5a --- /dev/null +++ b/core/include/vdm/io/sparse_file.hpp @@ -0,0 +1,80 @@ +// vdm/io/sparse_file.hpp — the one output file, written at absolute offsets. +// +// docs/04 §4: one file, opened once, O_WRONLY; posix_fallocate the full size up front; +// each segment pwrite()s at its own offset so there is no reassembly pass; fadvise +// DONTNEED on written ranges; fdatasync on a timer, never per write. +// +// This header compiles standalone. + +#ifndef VDM_IO_SPARSE_FILE_HPP +#define VDM_IO_SPARSE_FILE_HPP + +#include +#include +#include + +#include "vdm/util/bytes.hpp" +#include "vdm/util/result.hpp" + +namespace vdm::io { + +class SparseFile { + public: + struct OpenOptions { + std::uint64_t total_size = 0; // full final size; 0 = unknown (chunked transfer) + bool preallocate = true; // posix_fallocate; falls back to ftruncate + bool truncate_existing = false; // start fresh (true) vs. resume into an existing + // part file (false) + }; + + SparseFile() = default; + ~SparseFile(); + + SparseFile(SparseFile &&) noexcept; + SparseFile &operator=(SparseFile &&) noexcept; + SparseFile(const SparseFile &) = delete; + SparseFile &operator=(const SparseFile &) = delete; + + // Open `path` for writing at absolute offsets, creating it if needed. With + // `preallocate` and a known `total_size`, posix_fallocate the whole file (contiguous + // extents, no ENOSPC surprise at 99%). EOPNOTSUPP/ENOSYS (tmpfs, some network FS) + // falls back to ftruncate — sparse, no extent reservation — and is reported by + // preallocated(). + [[nodiscard]] Result open(std::string_view path, const OpenOptions &opts); + [[nodiscard]] Result open(std::string_view path); // default OpenOptions + + [[nodiscard]] bool is_open() const noexcept { return fd_ >= 0; } + [[nodiscard]] bool preallocated() const noexcept { return preallocated_; } + [[nodiscard]] std::string_view path() const noexcept { return path_; } + + // pwrite the whole span at `offset`, looping over short writes and retrying EINTR. + // Safe to call concurrently with other write_at()/sync() on the same object as long + // as the byte ranges do not overlap — POSIX guarantees each pwrite is atomic for a + // regular file, so no lock is taken on the hot path. + [[nodiscard]] Result write_at(std::uint64_t offset, ConstByteSpan data); + + // fdatasync. Call on a timer (default 5 s) and on pause — never per write (docs/04 §4). + [[nodiscard]] Result sync(); + + // POSIX_FADV_DONTNEED on [offset, offset+len): drop already-written pages from the + // page cache so a 40 GB ISO does not evict the user's working set. Most effective + // after sync(). Best-effort — failures are ignored. + void advise_dontneed(std::uint64_t offset, std::uint64_t len) noexcept; + + // ftruncate to `size`: give a chunked download its real size once known, or trim a + // preallocated tail that a steal/mirror never filled. + [[nodiscard]] Result resize(std::uint64_t size); + + [[nodiscard]] Result close(); + + private: + void reset() noexcept; + + int fd_ = -1; + bool preallocated_ = false; + std::string path_; +}; + +} // namespace vdm::io + +#endif // VDM_IO_SPARSE_FILE_HPP diff --git a/core/include/vdm/io/write_buffer.hpp b/core/include/vdm/io/write_buffer.hpp new file mode 100644 index 0000000..e499ef5 --- /dev/null +++ b/core/include/vdm/io/write_buffer.hpp @@ -0,0 +1,69 @@ +// vdm/io/write_buffer.hpp — per-segment accumulate-and-flush buffer. +// +// docs/04 §4: curl's write callback appends; the buffer is flushed with a single pwrite +// when full or when the segment ends. §8: NO ALLOCATION in the write-callback hot path — +// the buffer is preallocated at construction and append() only memcpys. +// +// Single-threaded. The owning download_task (stage 8) adds the disk-writer-thread handoff +// (double buffering) on top; this primitive takes no lock. +// +// This header compiles standalone. + +#ifndef VDM_IO_WRITE_BUFFER_HPP +#define VDM_IO_WRITE_BUFFER_HPP + +#include +#include +#include +#include + +#include "vdm/util/bytes.hpp" +#include "vdm/util/result.hpp" + +namespace vdm::io { + +class WriteBuffer { + public: + // Writes `span` durably at absolute file offset `offset`. Must not allocate + // (SparseFile::write_at doesn't). Return an error to abort the segment; WriteBuffer + // propagates it and keeps the un-flushed bytes so the caller can decide. + using FlushFn = std::function(std::uint64_t offset, ConstByteSpan span)>; + + // `capacity` is the effective per-segment buffer_bytes (already clamped by the + // segmenter against max_total_buffer_bytes). Must be > 0. + WriteBuffer(std::uint64_t start_offset, std::size_t capacity, FlushFn flush); + + WriteBuffer(WriteBuffer &&) noexcept = default; + WriteBuffer &operator=(WriteBuffer &&) noexcept = default; + WriteBuffer(const WriteBuffer &) = delete; + WriteBuffer &operator=(const WriteBuffer &) = delete; + + // Append body bytes. Flushes automatically each time the buffer fills. A chunk at + // least `capacity` bytes long, arriving when the buffer is empty, is written straight + // through (one extra flush call, still no allocation and no memcpy). + [[nodiscard]] Result append(ConstByteSpan span); + + // Flush whatever is buffered right now. Call at segment end and on pause. A no-op + // (success) when nothing is pending. + [[nodiscard]] Result flush(); + + [[nodiscard]] std::size_t capacity() const noexcept { return buf_.size(); } + [[nodiscard]] std::size_t pending() const noexcept { return len_; } + // File offset the next flush will write at (start + all bytes already flushed). + [[nodiscard]] std::uint64_t next_offset() const noexcept { return base_; } + // Total bytes handed to append() over this buffer's life. + [[nodiscard]] std::uint64_t total_appended() const noexcept { return appended_; } + + private: + Result flush_pending(); + + std::vector buf_; + std::size_t len_ = 0; // bytes currently in buf_ + std::uint64_t base_ = 0; // file offset of buf_[0] + std::uint64_t appended_ = 0; + FlushFn flush_; +}; + +} // namespace vdm::io + +#endif // VDM_IO_WRITE_BUFFER_HPP diff --git a/core/src/io/sparse_file.cpp b/core/src/io/sparse_file.cpp new file mode 100644 index 0000000..45e2175 --- /dev/null +++ b/core/src/io/sparse_file.cpp @@ -0,0 +1,181 @@ +// vdm/io/sparse_file.cpp + +#include "vdm/io/sparse_file.hpp" + +#include +#include + +#include +#include +#include + +namespace vdm::io { +namespace { + +Error errno_to_error(int e) noexcept { + switch (e) { + case ENOSPC: + case EDQUOT: + return Error::disk_full; + case EACCES: + case EPERM: + case EROFS: + return Error::permission_denied; + case ENOENT: + case ENOTDIR: + case EISDIR: + case ENAMETOOLONG: + case ELOOP: + return Error::path_rejected; + default: + return Error::io_error; + } +} + +ErrorInfo sys_error(std::string_view what, int e) { + return ErrorInfo(errno_to_error(e), std::string(what) + ": " + std::strerror(e)); +} + +} // namespace + +SparseFile::~SparseFile() { + if (fd_ >= 0) + ::close(fd_); +} + +SparseFile::SparseFile(SparseFile &&o) noexcept + : fd_(std::exchange(o.fd_, -1)), + preallocated_(std::exchange(o.preallocated_, false)), + path_(std::move(o.path_)) {} + +SparseFile &SparseFile::operator=(SparseFile &&o) noexcept { + if (this != &o) { + if (fd_ >= 0) + ::close(fd_); + fd_ = std::exchange(o.fd_, -1); + preallocated_ = std::exchange(o.preallocated_, false); + path_ = std::move(o.path_); + } + return *this; +} + +void SparseFile::reset() noexcept { + fd_ = -1; + preallocated_ = false; + path_.clear(); +} + +Result SparseFile::open(std::string_view path) { + return open(path, OpenOptions{}); +} + +Result SparseFile::open(std::string_view path, const OpenOptions &opts) { + if (fd_ >= 0) + return ErrorInfo(Error::internal, "SparseFile already open"); + + std::string p(path); + int flags = O_WRONLY | O_CREAT | O_CLOEXEC; + if (opts.truncate_existing) + flags |= O_TRUNC; + + int fd = ::open(p.c_str(), flags, 0644); + if (fd < 0) + return sys_error("open " + p, errno); + + bool prealloc = false; + if (opts.total_size > 0) { + if (opts.preallocate) { + // posix_fallocate returns the error number directly and does not set errno. + int rc = ::posix_fallocate(fd, 0, static_cast(opts.total_size)); + if (rc == 0) { + prealloc = true; + } else if (rc == EOPNOTSUPP || rc == ENOSYS || rc == EINVAL) { + if (::ftruncate(fd, static_cast(opts.total_size)) != 0) { + int e = errno; + ::close(fd); + return sys_error("ftruncate " + p, e); + } + } else { + ::close(fd); + return sys_error("posix_fallocate " + p, rc); + } + } else if (!opts.truncate_existing) { + // Resuming: make sure the file is at least total_size so pwrite offsets land. + if (::ftruncate(fd, static_cast(opts.total_size)) != 0) { + int e = errno; + ::close(fd); + return sys_error("ftruncate " + p, e); + } + } + } + + fd_ = fd; + preallocated_ = prealloc; + path_ = std::move(p); + return ok(); +} + +Result SparseFile::write_at(std::uint64_t offset, ConstByteSpan data) { + if (fd_ < 0) + return ErrorInfo(Error::internal, "write_at on a closed SparseFile"); + + const std::byte *p = data.data(); + std::size_t remaining = data.size(); + off_t pos = static_cast(offset); + + while (remaining > 0) { + ssize_t n = ::pwrite(fd_, p, remaining, pos); + if (n < 0) { + if (errno == EINTR) + continue; + return sys_error("pwrite", errno); + } + if (n == 0) + return ErrorInfo(Error::io_error, "pwrite returned 0"); + p += n; + pos += n; + remaining -= static_cast(n); + } + return ok(); +} + +Result SparseFile::sync() { + if (fd_ < 0) + return ErrorInfo(Error::internal, "sync on a closed SparseFile"); + while (::fdatasync(fd_) != 0) { + if (errno == EINTR) + continue; + return sys_error("fdatasync", errno); + } + return ok(); +} + +void SparseFile::advise_dontneed(std::uint64_t offset, std::uint64_t len) noexcept { + if (fd_ < 0 || len == 0) + return; + ::posix_fadvise(fd_, static_cast(offset), static_cast(len), POSIX_FADV_DONTNEED); +} + +Result SparseFile::resize(std::uint64_t size) { + if (fd_ < 0) + return ErrorInfo(Error::internal, "resize on a closed SparseFile"); + while (::ftruncate(fd_, static_cast(size)) != 0) { + if (errno == EINTR) + continue; + return sys_error("ftruncate", errno); + } + return ok(); +} + +Result SparseFile::close() { + if (fd_ < 0) + return ok(); + int fd = std::exchange(fd_, -1); + int rc = ::close(fd); + reset(); + if (rc != 0) + return sys_error("close", errno); + return ok(); +} + +} // namespace vdm::io diff --git a/core/src/io/write_buffer.cpp b/core/src/io/write_buffer.cpp new file mode 100644 index 0000000..0a407d0 --- /dev/null +++ b/core/src/io/write_buffer.cpp @@ -0,0 +1,57 @@ +// vdm/io/write_buffer.cpp + +#include "vdm/io/write_buffer.hpp" + +#include +#include +#include +#include + +namespace vdm::io { + +WriteBuffer::WriteBuffer(std::uint64_t start_offset, std::size_t capacity, FlushFn flush) + : buf_(capacity), base_(start_offset), flush_(std::move(flush)) { + assert(capacity > 0 && "WriteBuffer capacity must be > 0"); +} + +Result WriteBuffer::flush_pending() { + if (len_ == 0) + return ok(); + VDM_TRY(flush_(base_, ConstByteSpan(buf_.data(), len_))); + base_ += len_; + len_ = 0; + return ok(); +} + +Result WriteBuffer::append(ConstByteSpan span) { + // On an error return, next_offset() still reflects exactly what is durable; buffered + // (non-durable) bytes and any unconsumed tail of `span` are the caller's to abandon — + // the segment aborts and resumes/restarts from next_offset(). + while (!span.empty()) { + // Buffer empty and the incoming chunk fills at least a whole buffer: write it + // straight through — no memcpy, no allocation, just an extra flush call. + if (len_ == 0 && span.size() >= buf_.size()) { + VDM_TRY(flush_(base_, span)); + base_ += span.size(); + appended_ += span.size(); + return ok(); + } + + const std::size_t room = buf_.size() - len_; + const std::size_t n = std::min(span.size(), room); + std::memcpy(buf_.data() + len_, span.data(), n); + len_ += n; + appended_ += n; + span = span.subspan(n); + + if (len_ == buf_.size()) + VDM_TRY(flush_pending()); + } + return ok(); +} + +Result WriteBuffer::flush() { + return flush_pending(); +} + +} // namespace vdm::io diff --git a/core/tests/CMakeLists.txt b/core/tests/CMakeLists.txt index 6a465ac..e215eab 100644 --- a/core/tests/CMakeLists.txt +++ b/core/tests/CMakeLists.txt @@ -25,6 +25,8 @@ vdm_add_test(veloxcore_log_test util/log_test.cpp) # in the tree yet (lanes merge independently). vdm_add_test(veloxcore_content_disposition_test net/content_disposition_test.cpp) vdm_add_test(veloxcore_url_test net/url_test.cpp) +vdm_add_test(veloxcore_sparse_file_test io/sparse_file_test.cpp) +vdm_add_test(veloxcore_write_buffer_test io/write_buffer_test.cpp) set(_testserver ${CMAKE_SOURCE_DIR}/tools/testserver/testserver.py) foreach(net_it http_client probe) diff --git a/core/tests/io/sparse_file_test.cpp b/core/tests/io/sparse_file_test.cpp new file mode 100644 index 0000000..1a4b469 --- /dev/null +++ b/core/tests/io/sparse_file_test.cpp @@ -0,0 +1,181 @@ +#include "vdm/io/sparse_file.hpp" + +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#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 read_all(const std::string &path) { + int fd = ::open(path.c_str(), O_RDONLY); + if (fd < 0) + return {}; + std::vector out; + std::array 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(end); +} + +vdm::ConstByteSpan bytes(const char *s) { + return {reinterpret_cast(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 ts; + for (int s = 0; s < kSegs; ++s) { + ts.emplace_back([&, s] { + std::vector chunk(kSeg, static_cast('A' + s)); + for (std::size_t off = 0; off < kSeg; off += 4096) { + auto r = f.write_at(static_cast(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('A' + s)) + ok = false; + VT_CHECK(ok); + } + f.close().value(); +} diff --git a/core/tests/io/write_buffer_test.cpp b/core/tests/io/write_buffer_test.cpp new file mode 100644 index 0000000..093f821 --- /dev/null +++ b/core/tests/io/write_buffer_test.cpp @@ -0,0 +1,202 @@ +#include "vdm/io/write_buffer.hpp" + +#include +#include +#include +#include +#include +#include + +#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 g_alloc_calls{0}; +std::atomic 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 data; // pre-reserved + std::vector offs; // pre-reserved + std::vector lens; + bool fail_next = false; + + WriteBuffer::FlushFn fn() { + return [this](std::uint64_t off, ConstByteSpan s) -> Result { + 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(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(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(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 total{0}; + auto flush = [&total](std::uint64_t, ConstByteSpan s) -> Result { + 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 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); +} diff --git a/core/tests/support/vtest.hpp b/core/tests/support/vtest.hpp index 0aea90f..fa48322 100644 --- a/core/tests/support/vtest.hpp +++ b/core/tests/support/vtest.hpp @@ -129,11 +129,14 @@ inline int run_all() { ::vt::report(__FILE__, __LINE__, #COND, {}, /*fatal=*/true); \ } while (0) +// NOTE: operands are copied (auto, not auto&&). A test assertion must never outlive a +// temporary the expression returned a reference into — the copy makes that safe. All +// compared types here are cheap to copy. #define VT_CHECK_EQ(A, B) \ do { \ ::vt::stats().checks++; \ - auto &&_a = (A); \ - auto &&_b = (B); \ + auto _a = (A); \ + auto _b = (B); \ if (!(_a == _b)) \ ::vt::report(__FILE__, __LINE__, #A " == " #B, \ ::vt::show(_a) + " vs " + ::vt::show(_b), false); \ @@ -142,8 +145,8 @@ inline int run_all() { #define VT_CHECK_NE(A, B) \ do { \ ::vt::stats().checks++; \ - auto &&_a = (A); \ - auto &&_b = (B); \ + auto _a = (A); \ + auto _b = (B); \ if (!(_a != _b)) \ ::vt::report(__FILE__, __LINE__, #A " != " #B, \ ::vt::show(_a) + " vs " + ::vt::show(_b), false); \