diff --git a/meson.build b/meson.build index 4dedf6a..edbd5db 100644 --- a/meson.build +++ b/meson.build @@ -30,8 +30,22 @@ libcurl = dependency('libcurl') libsystemd = dependency('libsystemd') pthreads = dependency('threads') stdcppfs = cpp.find_library('stdc++fs') -gtest = dependency('gtest', required: false) -gmock = dependency('gmock', required: false) +gtest = declare_dependency( + dependencies: [ + dependency( + 'gtest', + version: '>=1.10.0', + required: get_option('unittests'), + disabler: true, + ), + dependency( + 'gtest_main', + required: get_option('unittests'), + disabler: true, + ), + ], +) +gmock = dependency('gmock', required: false, disabler: true) pod2man = find_program('pod2man') pkgconfig = find_program('pkg-config') @@ -55,7 +69,7 @@ libcommon = static_library( src/queue.hh '''.split(), ), - dependencies: [libpcre, libarchive, libcurl, pthreads, stdcppfs], + dependencies: [libpcre, libarchive, libcurl, libsystemd, pthreads, stdcppfs], install: false, ) @@ -185,30 +199,28 @@ install_data( install_dir: join_paths(get_option('datadir'), 'licenses/pkgfile'), ) -if gtest.found() and gmock.found() - gtest_main = static_library('gtest_main', 'src/test/gtest_main.cc') - - test( +test( + 'unit_tests', + executable( 'unit_tests', - executable( - 'unit_tests', - files( - ''' - src/db_test.cc - src/filter_test.cc - src/queue_test.cc - src/repo_test.cc - src/result_test.cc - '''.split(), - ), - link_with: [libcommon, gtest_main], - dependencies: [gmock, gtest, libpcre, stdcppfs], + files( + ''' + src/db_test.cc + src/filter_test.cc + src/queue_test.cc + src/repo_test.cc + src/result_test.cc + '''.split(), ), - protocol: 'gtest', - ) -else - message('Skipping unit tests, gtest or gmock not found') -endif + link_with: [libcommon], + dependencies: [gmock, gtest, libpcre, stdcppfs], + ), + env: [ + 'TZ=UTC', + 'LC_TIME=C', + ], + protocol: 'gtest', +) python = import('python') py3 = python.find_installation('python3') @@ -216,11 +228,11 @@ py3 = python.find_installation('python3') python_requirement = '>=3.7' if py3.found() and py3.language_version().version_compare(python_requirement) integration_tests = [ - 'tests/database.py', - 'tests/list.py', - 'tests/pkgfiled.py', - 'tests/search.py', - 'tests/update.py' + 'tests/database.py', + 'tests/list.py', + 'tests/pkgfiled.py', + 'tests/search.py', + 'tests/update.py', ] foreach input : integration_tests basename = input.split('/')[-1].split('.')[0] @@ -230,7 +242,7 @@ if py3.found() and py3.language_version().version_compare(python_requirement) py3, args: [join_paths(meson.project_source_root(), input)], env: ['PYTHONDONTWRITEBYTECODE=1'], - suite: 'integration' + suite: 'integration', ) endforeach else diff --git a/meson_options.txt b/meson_options.txt index 4575bd8..eb1177a 100644 --- a/meson_options.txt +++ b/meson_options.txt @@ -7,3 +7,6 @@ option('systemd_units', type : 'boolean', value : true, description : 'install systemd units to system unit directory') + +option('unittests', type : 'feature', value : 'auto', + description : 'Include unit tests in the build. Depends on gtest and gmock.') diff --git a/src/db_builder.cc b/src/db_builder.cc index 04e1279..f4dcba0 100644 --- a/src/db_builder.cc +++ b/src/db_builder.cc @@ -174,6 +174,23 @@ std::unique_ptr DbBuilder::FromArchive( bool DbBuilder::WriteToFile(const std::string& path, int64_t mtime) { const StringId repo_name_id = InternString(reponame_); + // The packed path table's 24-bit fields cap both how many distinct paths + // and how many distinct strings this repo can hold -- see + // kMaxPackedPathCount in db_format.hh. Checked after interning the repo + // name above (its own last possible addition to strings_), so the bound + // covers everything that's about to be serialized. Real repos are nowhere + // close (Arch's `extra` sits at roughly 2.5x headroom under this as of + // writing), so this is a hard error rather than something worth degrading + // gracefully for. + if (paths_.size() > kMaxPackedPathCount || + strings_.size() > kMaxPackedPathCount) { + std::cerr << std::format( + "error: repo exceeds the packed path-table capacity ({} paths, {} " + "strings, limit {})\n", + paths_.size(), strings_.size(), kMaxPackedPathCount); + return false; + } + // Sort packages by name, and remap every reference to a package's original // (insertion-order) index to its new, sorted PkgId so that list mode can // binary search the package table. @@ -217,7 +234,10 @@ bool DbBuilder::WriteToFile(const std::string& path, int64_t mtime) { [&](StringId a, StringId b) { return strings_[a] < strings_[b]; }); std::vector basename_index; - std::vector postings_pool; + // Delta+varint-encoded pooled postings, keyed by byte offset rather than + // index -- see the format comment on BasenameEntry and + // EncodePostingsDelta in db_format.hh. + std::string postings_blob; basename_index.reserve(basenames.size()); for (const StringId basename_id : basenames) { @@ -240,25 +260,36 @@ bool DbBuilder::WriteToFile(const std::string& path, int64_t mtime) { postings[0].path, }); } else { + // postings_blob.size() must fit in kPostingsStartMask's 31 bits, same + // as the old index-based offset did implicitly (a repo would need a + // ~2GB postings blob to hit this; nowhere close to any real repo). + if (postings_blob.size() > kPostingsStartMask) { + std::cerr << std::format("error: postings pool exceeds {} bytes\n", + kPostingsStartMask); + return false; + } basename_index.push_back(BasenameEntry{ basename_id, - static_cast(postings_pool.size()), + static_cast(postings_blob.size()), static_cast(postings.size()), }); - postings_pool.insert(postings_pool.end(), postings.begin(), - postings.end()); + EncodePostingsDelta(&postings_blob, postings); } } - // Byte pool + string table, in original StringId order. + // Byte pool + string table, in original StringId order. The string table + // stores only each string's starting offset into the byte pool; its + // length is implicit from the next entry's offset (or, for the last + // string, this trailing sentinel), since the byte pool below is built by + // pure concatenation in this same order. std::string byte_pool; - std::vector string_table; - string_table.reserve(strings_.size()); + std::vector string_table; + string_table.reserve(strings_.size() + 1); for (const auto& s : strings_) { - string_table.push_back(StringRef{static_cast(byte_pool.size()), - static_cast(s.size())}); + string_table.push_back(static_cast(byte_pool.size())); byte_pool += s; } + string_table.push_back(static_cast(byte_pool.size())); std::string buf; buf.resize(sizeof(Header), '\0'); @@ -275,12 +306,27 @@ bool DbBuilder::WriteToFile(const std::string& path, int64_t mtime) { AppendAligned(&buf, byte_pool.data(), byte_pool.size(), 8); header.string_table_offset = buf.size(); - header.string_table_count = string_table.size(); + // strings_.size(), not string_table.size() -- the table physically holds + // one extra trailing-sentinel entry (see the field's doc comment in + // db_format.hh), which string_table_count deliberately excludes. + header.string_table_count = strings_.size(); AppendAligned(&buf, string_table.data(), string_table.size(), 8); + // Pack the path trie into kPackedPathNodeSize-byte nodes (see + // db_format.hh); paths_.size() was already checked against + // kMaxPackedPathCount above. + std::string packed_paths; + packed_paths.reserve(paths_.size() * kPackedPathNodeSize); + for (const PathNode& node : paths_) { + uint8_t packed[kPackedPathNodeSize]; + PackPathNode24(packed, node.parent, node.name); + packed_paths.append(reinterpret_cast(packed), + kPackedPathNodeSize); + } + header.path_table_offset = buf.size(); header.path_table_count = paths_.size(); - AppendAligned(&buf, paths_.data(), paths_.size(), 8); + AppendAligned(&buf, packed_paths.data(), packed_paths.size(), 8); header.package_table_offset = buf.size(); header.package_table_count = package_table.size(); @@ -295,8 +341,10 @@ bool DbBuilder::WriteToFile(const std::string& path, int64_t mtime) { AppendAligned(&buf, basename_index.data(), basename_index.size(), 8); header.postings_offset = buf.size(); - header.postings_count = postings_pool.size(); - AppendAligned(&buf, postings_pool.data(), postings_pool.size(), 8); + // Byte length of the blob, not a Posting count -- see the field's doc + // comment in db_format.hh. + header.postings_count = postings_blob.size(); + AppendAligned(&buf, postings_blob.data(), postings_blob.size(), 8); buf.replace(0, sizeof(Header), reinterpret_cast(&header), sizeof(Header)); diff --git a/src/db_format.hh b/src/db_format.hh index 48141a6..7edbc08 100644 --- a/src/db_format.hh +++ b/src/db_format.hh @@ -2,21 +2,38 @@ // On-disk layout for the pkgfile repo database ("PFDB"). // -// A PFDB file is a single flat binary blob, mmap'd read-only and read -// directly as POD structs -- there is no decompression or deserialization -// step. All cross references are indices into one of the tables below rather -// than pointers, so the file is position-independent and can be mapped -// anywhere in the address space. +// A PFDB file is a single flat binary blob, mmap'd read-only. Most sections +// are read directly as POD structs with no decoding step. Two exceptions, +// both chosen because they're either always randomly accessed anyway (so +// narrowing the record instead of varint-encoding it keeps O(1) indexing) +// or read only as small, already-located slices (so a lightweight decode +// costs nothing beyond what the caller was already paying to iterate it): +// - the string table stores one u32 byte-pool offset per string rather +// than an {offset,length} pair -- length is implicit from the next +// entry's offset, since the byte pool is written by pure concatenation. +// - the path trie's nodes are packed into 6 bytes each (two 24-bit +// fields) rather than the natural 8, since a PathId/StringId +// comfortably fits 24 bits for any repo pkgfile realistically indexes. +// - the postings pool -- reached only after a binary search on the +// (still fixed-width) basename index, then read as one small bounded +// slice -- is delta+varint encoded rather than a flat Posting array. +// All cross references are indices (or, for the postings pool, byte +// offsets) into one of the tables below rather than pointers, so the file +// is position-independent and can be mapped anywhere in the address space. // -// Compaction comes from string interning: directory components that are -// shared by many files (e.g. "usr", "usr/lib") are stored exactly once in the -// path table, which is a "parent chain" trie -- a full path is a single -// PathId, and walking the `parent` field from that id back to the root yields -// the path's components in reverse order. +// Compaction also comes from string interning: directory components that +// are shared by many files (e.g. "usr", "usr/lib") are stored exactly once +// in the path table, which is a "parent chain" trie -- a full path is a +// single PathId, and walking the `parent` field from that id back to the +// root yields the path's components in reverse order. #include #include +#include +#include +#include #include +#include namespace pkgfile::db { @@ -48,16 +65,16 @@ inline constexpr uint32_t TagPath(PathId id, bool is_dir) { #pragma pack(push, 1) -// A reference to a byte range within the byte pool. -struct StringRef { - uint32_t offset; - uint32_t length; -}; -static_assert(sizeof(StringRef) == 8); - // One node in the path trie. `name` is a StringId for this component alone // (e.g. "bin", not "usr/bin"); `parent` is the PathId of the enclosing // directory, or kRootPath if this is a top-level component. +// +// This is the in-memory/API shape only -- DbBuilder builds the trie as a +// plain vector of these, and MappedRepo::PathNodeAt() unpacks into one to +// hand back to callers. On disk each node is packed into 6 bytes instead +// (see kPackedPathNodeSize / PackPathNode24 / UnpackPathNode24 below), so +// there's nothing in the mapping shaped like this struct to return a +// reference into. struct PathNode { uint32_t parent; uint32_t name; @@ -75,7 +92,10 @@ struct Package { static_assert(sizeof(Package) == 16); // One occurrence of a basename: which package it's in, and the full path -// (tagged with the directory bit) it occurred at. +// (tagged with the directory bit) it occurred at. This is the decoded, +// in-memory shape; on disk, pooled postings are delta+varint encoded (see +// EncodePostingsDelta/DecodePostingsDelta below) rather than stored as a +// flat array of these. struct Posting { uint32_t pkg; uint32_t path; // PathId, tagged with kDirBit @@ -88,8 +108,9 @@ static_assert(sizeof(Posting) == 8); // of them would spend a whole extra Posting (and a pointer chase to reach // it) on data that already fits in the two spare uint32_ts an entry has // lying around. So: `postings_start`'s high bit distinguishes two encodings. -// - clear (the common range case): postings_start/postings_count are a -// [start, start + count) slice into the postings pool, as usual. +// - clear (the common range case): `postings_start` is a byte offset into +// the postings pool's delta+varint blob, where `postings_count` entries +// are encoded starting there (see DecodePostingsDelta). // - set (the single-occurrence case): this basename has exactly one // occurrence, inlined directly into the entry -- postings_start (masked // with kPostingsStartMask) is its PkgId, and postings_count holds its @@ -122,6 +143,189 @@ inline constexpr uint32_t PostingCountOf(const BasenameEntry& entry) { return HasInlinePosting(entry) ? 1 : entry.postings_count; } +#pragma pack(pop) + +// -- Path table: 6-byte packed nodes -- +// +// Each PathNode is packed on disk into 6 bytes (three little-endian bytes +// each for `parent` and `name`) instead of the natural 8, since a PathId or +// StringId comfortably fits in 24 bits for any repo pkgfile will +// realistically see (Arch's `extra` has ~6.7M distinct paths and ~2.8M +// distinct strings as of writing -- roughly 2.5x headroom under the 24-bit +// ceiling below). The packed layout is fixed-width, so PathNodeAt() stays +// O(1): index by id, unpack. +inline constexpr size_t kPackedPathNodeSize = 6; + +// The packed `parent` field's "no parent" marker (the 24-bit analog of +// kRootPath), and the shared capacity ceiling DbBuilder::WriteToFile() +// enforces for both the path table (whose 24-bit `parent` field reserves +// this value as a sentinel, so the table itself may hold at most this many +// nodes) and the string table (whose 24-bit `name` field has no reserved +// value and could technically hold one more, but sharing one ceiling is +// simpler to reason about than tracking two). +inline constexpr uint32_t kMaxPackedPathCount = 0x00FF'FFFFu; // 16,777,215 + +// Packs/unpacks via two fixed-size memcpy calls (a uint32_t covering bytes +// [0,4), a uint16_t covering bytes [4,6)) rather than six individual byte +// loads/shifts. This is the standard type-punning-free way to do an +// unaligned load/store in C++ -- no UB, and compilers reliably fold each +// memcpy into a single unaligned move instruction -- which matters here +// because PathNodeAt() (below) calls this unconditionally on every +// ResolvePathInto(), including cache hits: a full-repo scan calls it once +// per file (millions of times for a large repo), so this is squarely on +// the hot path a glob/regex query drives. Values are packed in native byte +// order, same as every other multi-byte field in this format (see the +// byte_order_guard check in MappedRepo::Open) -- this isn't attempting a +// portable on-disk byte order, just an unaligned read/write of whatever +// order native uint32_t/uint16_t already use. +inline void PackPathNode24(uint8_t out[kPackedPathNodeSize], uint32_t parent, + uint32_t name) { + const uint32_t packed_parent = + (parent == kRootPath) ? kMaxPackedPathCount : parent; + const uint32_t lo4 = packed_parent | ((name & 0xFFu) << 24); + memcpy(out, &lo4, sizeof(lo4)); + const uint16_t hi2 = static_cast(name >> 8); + memcpy(out + sizeof(lo4), &hi2, sizeof(hi2)); +} + +inline PathNode UnpackPathNode24(const uint8_t in[kPackedPathNodeSize]) { + uint32_t lo4; + memcpy(&lo4, in, sizeof(lo4)); + uint16_t hi2; + memcpy(&hi2, in + sizeof(lo4), sizeof(hi2)); + + const uint32_t packed_parent = lo4 & 0x00FF'FFFFu; + const uint32_t name = (lo4 >> 24) | (static_cast(hi2) << 8); + return PathNode{ + packed_parent == kMaxPackedPathCount ? kRootPath : packed_parent, name}; +} + +// -- Postings pool: delta + varint -- +// +// Postings for a given basename are always written sorted ascending by +// (pkg, path) (see DbBuilder::WriteToFile), and the pool is only ever read +// as one small bounded slice at a time -- located by a binary search on the +// (fixed-width) basename index, never itself binary-searched -- so +// variable-length encoding here costs nothing beyond decoding the handful +// of postings a lookup actually matched (a real repo averages ~8 per +// pooled basename). +// +// Standard unsigned LEB128 varints. Each posting is two varints: +// - pkg delta from the previous posting's pkg (or from 0, for the first +// posting in a slice). +// - if that pkg is unchanged from the previous posting, the path delta +// from the previous posting's path; otherwise the path itself, +// absolute (a path from a different package isn't necessarily anywhere +// near the last one, so a delta would just as often inflate as shrink). + +inline void AppendVarint(std::string* buf, uint64_t value) { + while (value >= 0x80) { + buf->push_back(static_cast((value & 0x7F) | 0x80)); + value >>= 7; + } + buf->push_back(static_cast(value)); +} + +// Reads one varint from `data[*pos, size)`, advancing *pos past it on +// success. Returns false (leaving *pos and *out unspecified) on truncation +// or an encoding that would need more than 64 bits -- both are signs of a +// corrupt file, since every varint this format ever writes fits in a +// handful of bytes. +inline bool ReadVarint(const uint8_t* data, size_t size, size_t* pos, + uint64_t* out) { + uint64_t result = 0; + int shift = 0; + for (;;) { + if (*pos >= size || shift >= 64) { + return false; + } + const uint8_t byte = data[(*pos)++]; + result |= static_cast(byte & 0x7F) << shift; + if ((byte & 0x80) == 0) { + *out = result; + return true; + } + shift += 7; + } +} + +// Appends `postings` (already sorted ascending by (pkg, path)) to `buf` in +// the delta+varint wire format described above. +inline void EncodePostingsDelta(std::string* buf, + std::span postings) { + uint32_t prev_pkg = 0; + uint32_t prev_path = 0; + for (const Posting& p : postings) { + AppendVarint(buf, static_cast(p.pkg) - prev_pkg); + if (p.pkg == prev_pkg) { + AppendVarint(buf, static_cast(p.path) - prev_path); + } else { + AppendVarint(buf, p.path); + } + prev_pkg = p.pkg; + prev_path = p.path; + } +} + +// Decodes `count` postings starting at byte `start` of `blob[0, blob_size)` +// (the postings pool) into `*out` (cleared first, then filled in order). +// Returns false -- leaving `*out` in an unspecified state -- if `start` is +// out of range, `count` couldn't possibly fit in the bytes remaining (each +// posting needs at least 2 bytes, so this also bounds a corrupt/hostile +// count before it can drive an oversized allocation), a varint is +// truncated, or a decoded pkg/path would overflow 32 bits. Callers must +// check the return value rather than assume decoded content is complete; +// this only validates the encoding is well-formed, not that the decoded +// pkg/path values are in range for this repo's tables -- that's on the +// caller, same as for the inline-posting case. +inline bool DecodePostingsDelta(const uint8_t* blob, uint64_t blob_size, + uint64_t start, uint32_t count, + std::vector* out) { + out->clear(); + if (start > blob_size) { + return false; + } + const size_t size = static_cast(blob_size); + size_t pos = static_cast(start); + + const uint64_t max_possible_count = (size - pos) / 2; + if (count > max_possible_count) { + return false; + } + out->reserve(count); + + uint32_t prev_pkg = 0; + uint32_t prev_path = 0; + for (uint32_t i = 0; i < count; ++i) { + uint64_t pkg_delta; + if (!ReadVarint(blob, size, &pos, &pkg_delta)) { + return false; + } + const uint64_t pkg = static_cast(prev_pkg) + pkg_delta; + if (pkg > UINT32_MAX) { + return false; + } + + uint64_t second; + if (!ReadVarint(blob, size, &pos, &second)) { + return false; + } + const uint64_t path = + (pkg == prev_pkg) ? static_cast(prev_path) + second : second; + if (path > UINT32_MAX) { + return false; + } + + out->push_back( + Posting{static_cast(pkg), static_cast(path)}); + prev_pkg = static_cast(pkg); + prev_path = static_cast(path); + } + return true; +} + +#pragma pack(push, 1) + struct Header { char magic[4]; // "PFDB" uint32_t version; @@ -132,18 +336,26 @@ struct Header { uint32_t package_count; uint64_t byte_pool_offset, byte_pool_size; + // string_table_count is the number of distinct strings; the table itself + // physically holds string_table_count+1 uint32_t byte-pool offsets (one + // past the last string's start, as a sentinel), so that a string's length + // is always offset[id+1]-offset[id] without a separate length field. uint64_t string_table_offset, string_table_count; + // path_table_count is measured in kPackedPathNodeSize-byte packed nodes, + // not sizeof(PathNode). uint64_t path_table_offset, path_table_count; uint64_t package_table_offset, package_table_count; uint64_t package_files_offset, package_files_count; uint64_t basename_index_offset, basename_index_count; + // postings_count is the byte length of the delta+varint-encoded postings + // blob at postings_offset, not a count of Posting entries -- each + // BasenameEntry carries its own decode count (see PostingCountOf). uint64_t postings_offset, postings_count; }; static_assert(sizeof(Header) % 8 == 0); #pragma pack(pop) -static_assert(std::is_trivially_copyable_v); static_assert(std::is_trivially_copyable_v); static_assert(std::is_trivially_copyable_v); static_assert(std::is_trivially_copyable_v); @@ -151,7 +363,7 @@ static_assert(std::is_trivially_copyable_v); static_assert(std::is_trivially_copyable_v
); inline constexpr char kMagic[4] = {'P', 'F', 'D', 'B'}; -inline constexpr uint32_t kVersion = 1; +inline constexpr uint32_t kVersion = 2; inline constexpr uint32_t kByteOrderGuard = 0x01020304u; // Rounds `n` up to the next multiple of `align` (align must be a power of 2). diff --git a/src/db_test.cc b/src/db_test.cc index bf1f46c..a0b8f5c 100644 --- a/src/db_test.cc +++ b/src/db_test.cc @@ -97,8 +97,10 @@ TEST_F(DbRoundTripTest, RoundTripsPackagesAndFiles) { const auto* basename_entry = repo->FindBasename("bash"); ASSERT_NE(basename_entry, nullptr); EXPECT_FALSE(HasInlinePosting(*basename_entry)); - Posting bash_scratch; - const auto bash_postings = repo->PostingsFor(*basename_entry, &bash_scratch); + Posting bash_single; + std::vector bash_scratch; + const auto bash_postings = + repo->PostingsFor(*basename_entry, &bash_single, &bash_scratch); ASSERT_EQ(bash_postings.size(), 2u); bool found_binary = false; for (const auto& posting : bash_postings) { @@ -115,9 +117,10 @@ TEST_F(DbRoundTripTest, RoundTripsPackagesAndFiles) { const auto* getfattr_entry = repo->FindBasename("getfattr"); ASSERT_NE(getfattr_entry, nullptr); EXPECT_TRUE(HasInlinePosting(*getfattr_entry)); - Posting getfattr_scratch; + Posting getfattr_single; + std::vector getfattr_scratch; const auto getfattr_postings = - repo->PostingsFor(*getfattr_entry, &getfattr_scratch); + repo->PostingsFor(*getfattr_entry, &getfattr_single, &getfattr_scratch); ASSERT_EQ(getfattr_postings.size(), 1u); EXPECT_EQ( repo->ResolveString(repo->packages()[getfattr_postings[0].pkg].name), @@ -128,8 +131,10 @@ TEST_F(DbRoundTripTest, RoundTripsPackagesAndFiles) { const auto* usr_entry = repo->FindBasename("usr"); ASSERT_NE(usr_entry, nullptr); EXPECT_FALSE(HasInlinePosting(*usr_entry)); - Posting usr_scratch; - const auto usr_postings = repo->PostingsFor(*usr_entry, &usr_scratch); + Posting usr_single; + std::vector usr_scratch; + const auto usr_postings = + repo->PostingsFor(*usr_entry, &usr_single, &usr_scratch); ASSERT_EQ(usr_postings.size(), 2u); for (const auto& posting : usr_postings) { EXPECT_TRUE(IsDirOf(posting.path)); @@ -297,9 +302,13 @@ TEST_F(MappedRepoCorruptionTest, ResolveStringRejectsOutOfRangeByteRange) { Header header; memcpy(&header, bytes.data(), sizeof(header)); - // Claim a byte range that reaches far past the (tiny) byte pool this repo - // actually has. - PokeAt(&bytes, header.string_table_offset, StringRef{0, 0xFFFFFFFFu}); + // The string table stores one u32 byte-pool offset per string, plus a + // trailing sentinel; string 0's length is offsets[1]-offsets[0]. Poke + // offsets[1] (immediately after the table's first entry) to claim a byte + // range that reaches far past the (tiny) byte pool this repo actually + // has. + PokeAt(&bytes, header.string_table_offset + sizeof(uint32_t), + uint32_t{0xFFFFFFFFu}); WriteWholeFile(path_, bytes); MappedRepo::OpenError error; @@ -337,11 +346,15 @@ TEST_F(MappedRepoCorruptionTest, PathNodeAtRejectsForwardParent) { // parent in any legitimately-built db, since the builder always appends a // parent before its children. Point it at a later node instead -- the // shape of reference that would otherwise send a trie walk into an - // infinite loop on a corrupt file. - PathNode original; - memcpy(&original, bytes.data() + header.path_table_offset, sizeof(original)); - PokeAt(&bytes, header.path_table_offset, - PathNode{/*parent=*/2, original.name}); + // infinite loop on a corrupt file. Nodes are packed into + // kPackedPathNodeSize raw bytes on disk (see db_format.hh), so pack/unpack + // by hand rather than through PokeAt, which assumes its + // argument's natural (8-byte) layout. + uint8_t packed[kPackedPathNodeSize]; + memcpy(packed, bytes.data() + header.path_table_offset, kPackedPathNodeSize); + const PathNode original = UnpackPathNode24(packed); + PackPathNode24(packed, /*parent=*/2, original.name); + memcpy(bytes.data() + header.path_table_offset, packed, kPackedPathNodeSize); WriteWholeFile(path_, bytes); MappedRepo::OpenError error; @@ -445,13 +458,18 @@ TEST_F(MappedRepoCorruptionTest, PostingsForRejectsCorruptPooledEntry) { Header header; memcpy(&header, bytes.data(), sizeof(header)); - // Corrupt the pkg field of this basename's first pooled Posting so it - // points past the package table. - const size_t posting_offset = - header.postings_offset + postings_start * sizeof(Posting); - Posting original; - memcpy(&original, bytes.data() + posting_offset, sizeof(original)); - PokeAt(&bytes, posting_offset, Posting{/*pkg=*/0xFFFFFFFFu, original.path}); + // The postings pool is delta+varint-encoded (see db_format.hh): + // postings_start is a byte offset, and the first posting's pkg delta is + // from 0. With only two packages here, that delta fits in a single byte + // with no continuation bit, so overwriting just that one byte with + // another single-byte (still no continuation bit) varint corrupts only + // this posting's decoded pkg, leaving the rest of the blob's byte + // alignment untouched. 100 is comfortably past this repo's two-entry + // package table. + const size_t posting_offset = header.postings_offset + postings_start; + ASSERT_LT(static_cast(bytes[posting_offset]), 0x80) + << "test assumption: first posting's pkg delta fits in one byte"; + bytes[posting_offset] = static_cast(100); WriteWholeFile(path_, bytes); MappedRepo::OpenError error; @@ -459,8 +477,9 @@ TEST_F(MappedRepoCorruptionTest, PostingsForRejectsCorruptPooledEntry) { ASSERT_NE(repo, nullptr); const BasenameEntry* entry = repo->basename_index().data() + entry_index; - Posting scratch; - EXPECT_TRUE(repo->PostingsFor(*entry, &scratch).empty()); + Posting single; + std::vector scratch; + EXPECT_TRUE(repo->PostingsFor(*entry, &single, &scratch).empty()); } TEST_F(MappedRepoCorruptionTest, PostingsForRejectsCorruptInlinePkg) { @@ -502,8 +521,9 @@ TEST_F(MappedRepoCorruptionTest, PostingsForRejectsCorruptInlinePkg) { const BasenameEntry* entry = repo->basename_index().data() + entry_index; ASSERT_TRUE(HasInlinePosting(*entry)); - Posting scratch; - EXPECT_TRUE(repo->PostingsFor(*entry, &scratch).empty()); + Posting single; + std::vector scratch; + EXPECT_TRUE(repo->PostingsFor(*entry, &single, &scratch).empty()); } } // namespace diff --git a/src/mapped_repo.cc b/src/mapped_repo.cc index 8a5056e..20f009d 100644 --- a/src/mapped_repo.cc +++ b/src/mapped_repo.cc @@ -9,19 +9,26 @@ namespace pkgfile::db { namespace { -// Returns true if [offset, offset+count*sizeof(T)) fits within a file of +// Returns true if [offset, offset+count*elem_size) fits within a file of // `file_size` bytes, guarding every mmap'd access below against a truncated -// or corrupt database file. -template -bool FitsWithin(uint64_t offset, uint64_t count, uint64_t file_size) { +// or corrupt database file. Takes an explicit element size rather than a +// template parameter so it also covers the packed path table, whose 6-byte +// on-disk element doesn't correspond to any builtin type. +bool FitsWithinBytes(uint64_t offset, uint64_t count, uint64_t elem_size, + uint64_t file_size) { if (offset > file_size) { return false; } - const uint64_t bytes = count * sizeof(T); - return bytes / sizeof(T) == count && // overflow check + const uint64_t bytes = count * elem_size; + return (elem_size == 0 || bytes / elem_size == count) && // overflow check file_size - offset >= bytes; } +template +bool FitsWithin(uint64_t offset, uint64_t count, uint64_t file_size) { + return FitsWithinBytes(offset, count, sizeof(T), file_size); +} + } // namespace // static @@ -65,19 +72,30 @@ std::unique_ptr MappedRepo::Open(const std::string& path, return nullptr; } + // The string table physically holds one more entry than + // string_table_count (a trailing sentinel offset -- see the field's doc + // comment in db_format.hh); guard the +1 explicitly rather than let a + // maximally-corrupt count (UINT64_MAX) wrap it back to something small + // enough to spuriously pass the FitsWithin check below. + if (header->string_table_count == UINT64_MAX) { + *error = OpenError::kTruncated; + return nullptr; + } + const uint64_t string_table_span_count = header->string_table_count + 1; + if (!FitsWithin(header->byte_pool_offset, header->byte_pool_size, file_size) || - !FitsWithin(header->string_table_offset, - header->string_table_count, file_size) || - !FitsWithin(header->path_table_offset, header->path_table_count, - file_size) || + !FitsWithin(header->string_table_offset, + string_table_span_count, file_size) || + !FitsWithinBytes(header->path_table_offset, header->path_table_count, + kPackedPathNodeSize, file_size) || !FitsWithin(header->package_table_offset, header->package_table_count, file_size) || !FitsWithin(header->package_files_offset, header->package_files_count, file_size) || !FitsWithin(header->basename_index_offset, header->basename_index_count, file_size) || - !FitsWithin(header->postings_offset, header->postings_count, + !FitsWithin(header->postings_offset, header->postings_count, file_size)) { *error = OpenError::kTruncated; return nullptr; @@ -100,15 +118,23 @@ std::string_view MappedRepo::ResolveString(StringId id) const { if (id >= header_->string_table_count) { return {}; } - const StringRef ref = PtrAt(header_->string_table_offset)[id]; - // offset/length come straight from the file; check as uint64_t so a - // corrupt pair summing past UINT32_MAX can't wrap back into range. - if (uint64_t{ref.offset} + ref.length > header_->byte_pool_size) { + // The table holds one extra trailing-sentinel entry beyond + // string_table_count (see db_format.hh), so offsets[id+1] is always a + // valid index here -- Open() already checked the table fits + // string_table_count+1 entries. + const uint32_t* offsets = PtrAt(header_->string_table_offset); + const uint32_t start = offsets[id]; + const uint32_t end = offsets[id + 1]; + // start/end come straight from the file, so a corrupt pair could claim a + // range outside the byte pool, or an end before its own start (which, + // read as a length below, would wrap around to a huge unsigned value). + if (start > header_->byte_pool_size || end > header_->byte_pool_size || + end < start) { return {}; } return {reinterpret_cast(PtrAt(header_->byte_pool_offset) + - ref.offset), - ref.length}; + start), + end - start}; } namespace { @@ -123,7 +149,7 @@ void MappedRepo::ResolvePathInto(uint32_t tagged_path, std::string* out, PathCache* cache) const { const bool is_dir = IsDirOf(tagged_path); const PathId leaf = PathIdOf(tagged_path); - const PathNode& leaf_node = PathNodeAt(leaf); + const PathNode leaf_node = PathNodeAt(leaf); // Fast path: the previous call in this cache resolved a file with the // same immediate parent. Every ancestor beyond that parent is therefore @@ -148,9 +174,12 @@ void MappedRepo::ResolvePathInto(uint32_t tagged_path, std::string* out, size_t length = is_dir ? 1 : 0; bool overflowed = false; - const PathNode* node = &leaf_node; + // PathNodeAt() unpacks into a temporary (see its declaration), so this + // walks by value rather than chasing a pointer the way a direct reference + // into the mapping would allow. + PathNode node = leaf_node; for (PathId id = leaf;;) { - const std::string_view name = ResolveString(node->name); + const std::string_view name = ResolveString(node.name); length += 1 + name.size(); if (depth < kMaxInlineDepth) { components[depth] = name; @@ -159,11 +188,11 @@ void MappedRepo::ResolvePathInto(uint32_t tagged_path, std::string* out, } ++depth; - id = node->parent; + id = node.parent; if (id == kRootPath) { break; } - node = &PathNodeAt(id); + node = PathNodeAt(id); } if (overflowed) { @@ -247,8 +276,9 @@ const Package* MappedRepo::FindPackageByName(std::string_view name) const { return &*iter; } -std::span MappedRepo::PostingsFor(const BasenameEntry& entry, - Posting* single) const { +std::span MappedRepo::PostingsFor( + const BasenameEntry& entry, Posting* single, + std::vector* scratch) const { if (HasInlinePosting(entry)) { const PkgId pkg = InlinePkgOf(entry); const PathId path = PathIdOf(InlineTaggedPathOf(entry)); @@ -260,19 +290,23 @@ std::span MappedRepo::PostingsFor(const BasenameEntry& entry, return {single, 1}; } - const uint64_t table_count = header_->postings_count; - const uint64_t start = std::min(entry.postings_start, table_count); - const uint64_t count = - std::min(entry.postings_count, table_count - start); - const Posting* postings = PtrAt(header_->postings_offset) + start; + // entry.postings_start is a byte offset into the blob (its high bit is + // clear here, since HasInlinePosting() was false), not a Posting index -- + // see the format comment on BasenameEntry in db_format.hh. + const uint64_t blob_size = header_->postings_count; + const uint64_t start = std::min(entry.postings_start, blob_size); + if (!DecodePostingsDelta(PtrAt(header_->postings_offset), blob_size, + start, entry.postings_count, scratch)) { + return {}; + } - for (uint64_t i = 0; i < count; ++i) { - if (postings[i].pkg >= header_->package_table_count || - PathIdOf(postings[i].path) >= header_->path_table_count) { + for (const Posting& p : *scratch) { + if (p.pkg >= header_->package_table_count || + PathIdOf(p.path) >= header_->path_table_count) { return {}; } } - return {postings, count}; + return *scratch; } const BasenameEntry* MappedRepo::FindBasename(std::string_view name) const { diff --git a/src/mapped_repo.hh b/src/mapped_repo.hh index 2f6c9c1..af13e1e 100644 --- a/src/mapped_repo.hh +++ b/src/mapped_repo.hh @@ -6,6 +6,7 @@ #include #include #include +#include #include "archive_io.hh" #include "db_format.hh" @@ -99,12 +100,19 @@ class MappedRepo { // instead of indexing out of the mapping. The sentinel's own `parent` is // kRootPath, so a trie walk that hits it terminates immediately rather // than chasing a bad or cyclic reference any further. - const PathNode& PathNodeAt(PathId id) const { + // + // Returned by value: the on-disk path table packs each node into 6 bytes + // (see kPackedPathNodeSize in db_format.hh), so there's nothing in the + // mapping shaped like a PathNode to hand back a reference into -- this + // unpacks into a temporary instead. + PathNode PathNodeAt(PathId id) const { static constexpr PathNode kInvalid{kRootPath, 0}; if (id >= header_->path_table_count) { return kInvalid; } - const PathNode& node = PtrAt(header_->path_table_offset)[id]; + const uint8_t* packed = PtrAt(header_->path_table_offset) + + static_cast(id) * kPackedPathNodeSize; + const PathNode node = UnpackPathNode24(packed); if (node.parent != kRootPath && node.parent >= id) { return kInvalid; } @@ -121,6 +129,11 @@ class MappedRepo { size_t string_count() const { return header_->string_table_count; } uint64_t byte_pool_size() const { return header_->byte_pool_size; } + // The byte length of the delta+varint-encoded postings pool (see + // db_format.hh) -- not a Posting count. Exposed for introspection tools + // (see pfdb_dump.cc). + uint64_t postings_blob_size() const { return header_->postings_count; } + std::span packages() const { return {PtrAt(header_->package_table_offset), header_->package_table_count}; @@ -153,9 +166,12 @@ class MappedRepo { // Returns every (package, path) occurrence of `entry`'s basename. If // there's exactly one, it's inlined in `entry` itself (see db_format.hh) // and gets materialized into `*single`; the returned span points at - // `single` in that case, or into the shared postings pool otherwise, so - // callers must keep `single` alive for as long as the returned span is - // used. + // `single` in that case. Otherwise the pooled postings are decoded + // (delta+varint on disk -- see db_format.hh) into `*scratch` (cleared + // first) and the span points into that instead. Either way, callers must + // keep both `single` and `scratch` alive for as long as the returned span + // is used; pass the same `scratch` across a sequence of calls to reuse its + // buffer rather than paying for a fresh allocation each time. // // Bounds-checked, including the pkg/path each Posting carries -- callers // index packages() with a Posting's pkg field directly, so a corrupt @@ -165,7 +181,8 @@ class MappedRepo { // same (small) slice. Returns an empty span if anything in it is invalid, // rather than a partially-valid one. std::span PostingsFor(const BasenameEntry& entry, - Posting* single) const; + Posting* single, + std::vector* scratch) const; // Binary-searches the basename index for an exact, case-sensitive match. // Returns nullptr if no file in this repo has this basename. diff --git a/src/pfdb_dump.cc b/src/pfdb_dump.cc index a36e1d0..f7be9db 100644 --- a/src/pfdb_dump.cc +++ b/src/pfdb_dump.cc @@ -87,10 +87,13 @@ void CmdSummary(const MappedRepo& repo, uint64_t file_size, int64_t mtime) { std::cout << std::format(" {:<16} {:>10} {:>12}\n", "", "count", "bytes"); PrintSizeRow("byte pool", repo.string_count(), repo.byte_pool_size(), file_size); + // The string table physically holds one more entry than string_count() -- + // a trailing sentinel offset (see db_format.hh) -- at 4 bytes each (a raw + // offset, no separate length field). PrintSizeRow("string table", repo.string_count(), - repo.string_count() * sizeof(pkgfile::db::StringRef), file_size); + (repo.string_count() + 1) * sizeof(uint32_t), file_size); PrintSizeRow("path table", repo.path_count(), - repo.path_count() * sizeof(PathNode), file_size); + repo.path_count() * pkgfile::db::kPackedPathNodeSize, file_size); PrintSizeRow("package table", repo.packages().size(), repo.packages().size() * sizeof(Package), file_size); PrintSizeRow("package files", total_files, total_files * sizeof(uint32_t), @@ -98,7 +101,7 @@ void CmdSummary(const MappedRepo& repo, uint64_t file_size, int64_t mtime) { PrintSizeRow("basename index", repo.basename_index().size(), repo.basename_index().size() * sizeof(BasenameEntry), file_size); const uint64_t stored_postings = total_postings - inlined_postings; - PrintSizeRow("postings", stored_postings, stored_postings * sizeof(Posting), + PrintSizeRow("postings", stored_postings, repo.postings_blob_size(), file_size); std::cout << std::format( @@ -117,11 +120,12 @@ void CmdSummary(const MappedRepo& repo, uint64_t file_size, int64_t mtime) { if (repo.basename_index().size() > 0) { std::cout << std::format( "{} of {} basenames ({:.1f}%) have a single occurrence and are " - "inlined into the index, saving {} bytes that would otherwise sit " - "in the postings pool\n", + "inlined into the index, saving at least {} bytes that would " + "otherwise sit in the postings pool (each pooled posting needs a " + "minimum of 2 varint-encoded bytes; most need more)\n", inlined_postings, repo.basename_index().size(), 100.0 * inlined_postings / repo.basename_index().size(), - inlined_postings * sizeof(Posting)); + inlined_postings * 2); } } @@ -182,8 +186,9 @@ int CmdPostings(const MappedRepo& repo, std::string_view basename) { return 1; } - pkgfile::db::Posting scratch; - for (const auto& posting : repo.PostingsFor(*entry, &scratch)) { + pkgfile::db::Posting single; + std::vector scratch; + for (const auto& posting : repo.PostingsFor(*entry, &single, &scratch)) { const auto& pkg = repo.packages()[posting.pkg]; std::cout << std::format( "{} {:<40} {}\n", pkgfile::db::IsDirOf(posting.path) ? "d" : "f", diff --git a/src/pkgfile.cc b/src/pkgfile.cc index 2df2128..a3caa07 100644 --- a/src/pkgfile.cc +++ b/src/pkgfile.cc @@ -323,7 +323,7 @@ bool Pkgfile::PathMatches(const db::MappedRepo& repo, uint32_t tagged_path, return false; } - const db::PathNode& node = repo.PathNodeAt(id); + const db::PathNode node = repo.PathNodeAt(id); if (repo.ResolveString(node.name) != component) { return false; } @@ -375,8 +375,9 @@ void Pkgfile::SearchFullPathIndexed(const db::MappedRepo& repo, const filter::Bin is_bin(bins_); std::string resolved; - db::Posting scratch; - for (const auto& posting : repo.PostingsFor(*entry, &scratch)) { + db::Posting single; + std::vector scratch; + for (const auto& posting : repo.PostingsFor(*entry, &single, &scratch)) { if (db::IsDirOf(posting.path) != want_dir) { continue; } @@ -401,8 +402,10 @@ void Pkgfile::SearchExactIndexed(const db::MappedRepo& repo, std::string_view query, Result* result) { if (query.find('/') == query.npos) { if (const auto* entry = repo.FindBasename(query)) { - db::Posting scratch; - SearchBasenameIndexed(repo, repo.PostingsFor(*entry, &scratch), result); + db::Posting single; + std::vector scratch; + SearchBasenameIndexed(repo, repo.PostingsFor(*entry, &single, &scratch), + result); } } else { SearchFullPathIndexed(repo, query, result); @@ -429,6 +432,8 @@ void Pkgfile::ScanCaseInsensitive(const db::MappedRepo& repo, const filter::Bin is_bin(bins_); std::vector emitted(repo.packages().size(), false); std::string resolved; + db::Posting single; + std::vector scratch; for (const auto& entry : repo.basename_index()) { const std::string_view name = repo.ResolveString(entry.name); @@ -437,8 +442,7 @@ void Pkgfile::ScanCaseInsensitive(const db::MappedRepo& repo, continue; } - db::Posting scratch; - for (const auto& posting : repo.PostingsFor(entry, &scratch)) { + for (const auto& posting : repo.PostingsFor(entry, &single, &scratch)) { if (!options_.verbose && emitted[posting.pkg]) { continue; } @@ -705,8 +709,24 @@ int Pkgfile::RunList(const Database& database, std::string_view reponame, int Pkgfile::Run(const std::vector& args) { if (options_.mode & MODE_UPDATE) { - return Updater(options_.cachedir) - .Update(options_.cfgfile, options_.mode == MODE_UPDATE_FORCE); + sd_event* event = nullptr; + sd_event_new(&event); + + // Updater only registers work on `event`; it never runs the loop + // itself, so we own pumping it until the update's completion callback + // fires. + Updater updater(options_.cachedir); + int ret = 0; + updater.Update(event, options_.cfgfile, options_.mode == MODE_UPDATE_FORCE, + [&ret, event](int result) { + ret = result; + sd_event_exit(event, 0); + }); + + sd_event_loop(event); + sd_event_unref(event); + + return ret; } if (args.empty()) { diff --git a/src/pkgfiled.cc b/src/pkgfiled.cc index 30c60ef..c7a415d 100644 --- a/src/pkgfiled.cc +++ b/src/pkgfiled.cc @@ -1,5 +1,6 @@ #include #include +#include #include #include #include @@ -103,6 +104,16 @@ class Pkgfiled { &Pkgfiled::OnSignalEvent, this); sd_event_add_signal(sd_event_, &sigusr2_source_, SIGUSR2, &Pkgfiled::OnSignalEvent, this); + + // DB repacking can use a lot of RAM at peak, and glibc is stingy about + // releasing it. Schedule an idle priority event to trim the heap after + // higher priority events (repo and config processing) have been dispatched + // to drop idle RAM usage back down to kilobytes, instead of potentially + // hundreds of megabytes. + sd_event_add_defer(sd_event_, &malloc_trim_source_, &Pkgfiled::OnMallocTrim, + this); + sd_event_source_set_priority(malloc_trim_source_, SD_EVENT_PRIORITY_IDLE); + sd_event_source_set_enabled(malloc_trim_source_, SD_EVENT_OFF); } ~Pkgfiled() { @@ -111,6 +122,7 @@ class Pkgfiled { sd_event_source_unref(sigterm_source_); sd_event_source_unref(sigusr1_source_); sd_event_source_unref(sigusr2_source_); + sd_event_source_unref(malloc_trim_source_); sd_event_unref(sd_event_); sigprocmask(SIG_SETMASK, &saved_ss_, nullptr); @@ -177,10 +189,24 @@ class Pkgfiled { f.get(); } + ScheduleMallocTrim(); + return 0; } private: + // Arms the idle-priority malloc_trim defer source for a single dispatch. + // Re-arming while already armed is a no-op, so bursts of events in the same + // loop iteration still only trim once. + void ScheduleMallocTrim() { + sd_event_source_set_enabled(malloc_trim_source_, SD_EVENT_ONESHOT); + } + + static int OnMallocTrim(sd_event_source*, void*) { + malloc_trim(0); + return 0; + } + bool RepackRepo(const fs::path& changed_path) { auto repack = [&] { const std::string input_repo = watch_path_ / changed_path; @@ -228,6 +254,8 @@ class Pkgfiled { changed_path.filename().string(), dur.count()); } + ScheduleMallocTrim(); + return ok; } @@ -378,6 +406,7 @@ class Pkgfiled { sd_event_source* sigterm_source_; sd_event_source* sigusr1_source_; sd_event_source* sigusr2_source_; + sd_event_source* malloc_trim_source_; sigset_t saved_ss_{}; }; diff --git a/src/test/gtest_main.cc b/src/test/gtest_main.cc deleted file mode 100644 index 2f2d577..0000000 --- a/src/test/gtest_main.cc +++ /dev/null @@ -1,11 +0,0 @@ -#include - -#include "gtest/gtest.h" - -int main(int argc, char** argv) { - setenv("TZ", "UTC", 1); - setenv("LC_TIME", "C", 1); - - testing::InitGoogleTest(&argc, argv); - return RUN_ALL_TESTS(); -} diff --git a/src/update.cc b/src/update.cc index 93bf4f4..3871991 100644 --- a/src/update.cc +++ b/src/update.cc @@ -6,10 +6,12 @@ #include #include #include +#include +#include #include #include +#include -#include #include #include #include @@ -158,36 +160,9 @@ void PrintRepackSuccess(const std::string& reponame, double elapsed) { elapsed); } -int WaitForRepacking(std::vector* jobs, - bool show_message) { - if (show_message) { - int running = std::count_if( - jobs->begin(), jobs->end(), [](const pkgfile::DownloadJob& job) { - // The future won't be valid if the repo was up to date. - if (!job.worker.valid()) { - return false; - } - - return job.worker.wait_for(chrono::seconds::zero()) != - std::future_status::ready; - }); - - if (running > 0) { - std::cout << std::format( - ":: waiting for {} repo{} to finish repacking...\n", running, - running == 1 ? "" : "s"); - } - } - - return std::count_if(jobs->begin(), jobs->end(), - [](pkgfile::DownloadJob& job) { - return job.worker.valid() && !job.worker.get(); - }); -} - // curl's xfer info callback: reports download progress for one repo's -// transfer. Always runs on the main thread (inside curl_multi_perform), so -// no synchronization concerns on this side -- ProgressDisplay handles the +// transfer. Runs on the event loop thread (inside curl_multi_socket_action), +// so no synchronization concerns on this side -- ProgressDisplay handles the // rest, since repack progress does come from other threads. int XferInfoCallback(void* clientp, curl_off_t dltotal, curl_off_t dlnow, curl_off_t, curl_off_t) { @@ -198,6 +173,14 @@ int XferInfoCallback(void* clientp, curl_off_t dltotal, curl_off_t dlnow, return 0; } +// Aggregate state for one Update() call, kept alive by the shared_ptr +// captured in each of its jobs' on_done callbacks. +struct UpdateState { + int remaining; + int ret = 0; + std::set known_repos; +}; + } // namespace namespace pkgfile { @@ -208,7 +191,7 @@ DownloadJob::~DownloadJob() { } } -int Updater::DownloadQueueRequest(CURLM* multi, DownloadJob* job) { +int Updater::DownloadQueueRequest(DownloadJob* job) { if (job->curl == nullptr) { if (job->repo.servers.empty()) { std::cerr << std::format("error: no servers configured for repo {}\n", @@ -240,7 +223,7 @@ int Updater::DownloadQueueRequest(CURLM* multi, DownloadJob* job) { return -1; } } else { - curl_multi_remove_handle(multi, job->curl); + curl_multi_remove_handle(curl_multi_, job->curl); lseek(job->tmpfile.fd, 0, SEEK_SET); job->server_iter++; } @@ -268,7 +251,7 @@ int Updater::DownloadQueueRequest(CURLM* multi, DownloadJob* job) { } job->dl_time_start = now(); - curl_multi_add_handle(multi, job->curl); + curl_multi_add_handle(curl_multi_, job->curl); return 0; } @@ -346,131 +329,284 @@ void Updater::TidyCacheDir(const std::set& known_repos) { } } -void Updater::DownloadWaitLoop(CURLM* multi) { - int active_handles; - - do { - int nfd, rc = curl_multi_wait(multi, nullptr, 0, 1000, &nfd); - if (rc != CURLM_OK) { - std::cerr << std::format("error: curl_multi_wait failed ({})\n", rc); - break; - } - - if (nfd < 0) { - std::cerr << "error: poll error, possible network problem\n"; - break; - } +void Updater::CleanupCurl(DownloadJob* job) { + if (job->curl != nullptr) { + curl_multi_remove_handle(curl_multi_, job->curl); + curl_easy_cleanup(job->curl); + job->curl = nullptr; + } +} - rc = curl_multi_perform(multi, &active_handles); - if (rc != CURLM_OK) { - std::cerr << std::format("error: curl_multi_perform failed ({})\n", rc); - break; - } +void Updater::FinishJob(DownloadJob* job) { + CleanupCurl(job); - while (DownloadCheckComplete(multi, active_handles) == 0); - } while (active_handles > 0); -} + auto on_done = std::move(job->on_done); + Repo repo = std::move(job->repo); + DownloadResult result = job->dl_result; -int Updater::DownloadCheckComplete(CURLM* multi, int remaining) { - int msgs_left; + jobs_.remove_if([job](const DownloadJob& j) { return &j == job; }); + // `job` is now dangling; nothing below may touch it. - CURLMsg* msg = curl_multi_info_read(multi, &msgs_left); - if (msg == nullptr) { - return -1; + if (on_done) { + on_done(repo, result); } +} - if (msg->msg == CURLMSG_DONE) { - long uptodate, resp; - char* effective_url; - DownloadJob* job; - time_t remote_mtime; +void Updater::HandleDownloadComplete(DownloadJob* job, CURLMsg* msg) { + long uptodate = 0; + long resp = 0; + char* effective_url = nullptr; + time_t remote_mtime = 0; - curl_easy_getinfo(msg->easy_handle, CURLINFO_PRIVATE, &job); - curl_easy_getinfo(msg->easy_handle, CURLINFO_CONDITION_UNMET, &uptodate); - curl_easy_getinfo(msg->easy_handle, CURLINFO_RESPONSE_CODE, &resp); - curl_easy_getinfo(msg->easy_handle, CURLINFO_EFFECTIVE_URL, &effective_url); - curl_easy_getinfo(msg->easy_handle, CURLINFO_FILETIME_T, &remote_mtime); + curl_easy_getinfo(msg->easy_handle, CURLINFO_CONDITION_UNMET, &uptodate); + curl_easy_getinfo(msg->easy_handle, CURLINFO_RESPONSE_CODE, &resp); + curl_easy_getinfo(msg->easy_handle, CURLINFO_EFFECTIVE_URL, &effective_url); + curl_easy_getinfo(msg->easy_handle, CURLINFO_FILETIME_T, &remote_mtime); - if (uptodate) { - if (job->progress != nullptr) { - if (!job->progress->IsInteractive()) { - std::cout << std::format(" {} is up to date\n", job->repo.name); - } - job->progress->FinishDownload(job->progress_index, - ProgressDisplay::Stage::kSkipped); - job->progress->FinishRepack(job->progress_index, - ProgressDisplay::Stage::kSkipped); + if (uptodate) { + if (job->progress != nullptr) { + if (!job->progress->IsInteractive()) { + std::cout << std::format(" {} is up to date\n", job->repo.name); } - job->dl_result = DownloadResult::UPTODATE; - return 0; + job->progress->FinishDownload(job->progress_index, + ProgressDisplay::Stage::kSkipped); + job->progress->FinishRepack(job->progress_index, + ProgressDisplay::Stage::kSkipped); + } + job->dl_result = DownloadResult::UPTODATE; + FinishJob(job); + return; + } + + // was it a success? + if (msg->data.result != CURLE_OK || resp >= 400) { + if (*job->errmsg) { + std::cerr << std::format("warning: download failed: {}: {}\n", + effective_url, job->errmsg); + } else { + std::cerr << std::format("warning: download failed: {} [error {}]\n", + effective_url, resp); } - // was it a success? - if (msg->data.result != CURLE_OK || resp >= 400) { + if (DownloadQueueRequest(job) != 0) { + // No more servers left to retry: this repo is done for good. job->dl_result = DownloadResult::ERROR; - if (*job->errmsg) { - std::cerr << std::format("warning: download failed: {}: {}\n", - effective_url, job->errmsg); - } else { - std::cerr << std::format("warning: download failed: {} [error {}]\n", - effective_url, resp); - } - - const int r = DownloadQueueRequest(multi, job); - if (r != 0 && job->progress != nullptr) { - // No more servers left to retry: this repo is done for good. + if (job->progress != nullptr) { job->progress->FinishDownload(job->progress_index, ProgressDisplay::Stage::kFailed); job->progress->FinishRepack(job->progress_index, ProgressDisplay::Stage::kFailed); } - return r; + FinishJob(job); } + return; + } - job->tmpfile.size = lseek(job->tmpfile.fd, 0, SEEK_CUR); - lseek(job->tmpfile.fd, 0, SEEK_SET); + job->tmpfile.size = lseek(job->tmpfile.fd, 0, SEEK_CUR); + lseek(job->tmpfile.fd, 0, SEEK_SET); - struct timeval times[2] = { - {remote_mtime, 0}, - {remote_mtime, 0}, - }; - futimes(job->tmpfile.fd, times); + struct timeval times[2] = { + {remote_mtime, 0}, + {remote_mtime, 0}, + }; + futimes(job->tmpfile.fd, times); - if (job->progress != nullptr) { - const double elapsed = - chrono::duration(now() - job->dl_time_start).count(); - if (!job->progress->IsInteractive()) { - PrintDownloadSuccess(job, remaining, elapsed); - } - job->progress->FinishDownload(job->progress_index, - ProgressDisplay::Stage::kDone, elapsed); + if (job->progress != nullptr) { + const double elapsed = + chrono::duration(now() - job->dl_time_start).count(); + if (!job->progress->IsInteractive()) { + PrintDownloadSuccess(job, curl_running_handles_, elapsed); + } + job->progress->FinishDownload(job->progress_index, + ProgressDisplay::Stage::kDone, elapsed); + } + + // The curl transfer is done; release it now rather than waiting for the + // repack (which may take a while) to finish. + CleanupCurl(job); + StartRepack(job); +} + +void Updater::DrainCurlMessages() { + CURLMsg* msg; + int msgs_left; + while ((msg = curl_multi_info_read(curl_multi_, &msgs_left)) != nullptr) { + if (msg->msg != CURLMSG_DONE) { + continue; + } + DownloadJob* job = nullptr; + curl_easy_getinfo(msg->easy_handle, CURLINFO_PRIVATE, &job); + HandleDownloadComplete(job, msg); + } +} + +void Updater::StartRepack(DownloadJob* job) { + job->worker = std::async(std::launch::async, [this, job] { + const bool ok = RepackRepoData(job); + // Wake the event loop so it reaps this job's future without polling. + // The eventfd counter, not the write count, carries the notification, + // so a failed write here (ENOMEM-class only; the fd is always valid) + // just costs a poll-free wakeup we don't get -- nothing to recover. + uint64_t one = 1; + (void)!write(repack_eventfd_, &one, sizeof(one)); + return ok; + }); +} + +void Updater::ReapFinishedRepacks() { + // Snapshot which jobs are ready before finishing any of them: FinishJob() + // erases from jobs_, which would invalidate an in-progress iteration. + std::vector ready; + for (auto& job : jobs_) { + if (job.worker.valid() && job.worker.wait_for(chrono::seconds::zero()) == + std::future_status::ready) { + ready.push_back(&job); + } + } + + for (DownloadJob* job : ready) { + const bool repack_ok = job->worker.get(); + job->dl_result = repack_ok ? DownloadResult::OK : DownloadResult::ERROR; + FinishJob(job); + } +} + +int Updater::SocketCallback(CURL*, curl_socket_t s, int action, void* userp, + void* socketp) { + auto* self = static_cast(userp); + auto* io_source = static_cast(socketp); + + if (action == CURL_POLL_REMOVE) { + if (io_source != nullptr) { + sd_event_source_unref(io_source); + curl_multi_assign(self->curl_multi_, s, nullptr); + } + return 0; + } + + uint32_t events = 0; + if (action == CURL_POLL_IN || action == CURL_POLL_INOUT) { + events |= EPOLLIN; + } + if (action == CURL_POLL_OUT || action == CURL_POLL_INOUT) { + events |= EPOLLOUT; + } + + if (io_source == nullptr) { + sd_event_source* source = nullptr; + sd_event_add_io(self->event_, &source, s, events, &Updater::OnSocketReady, + self); + curl_multi_assign(self->curl_multi_, s, source); + } else { + sd_event_source_set_io_events(io_source, events); + } + + return 0; +} + +int Updater::OnSocketReady(sd_event_source*, int fd, uint32_t revents, + void* userdata) { + auto* self = static_cast(userdata); + + int ev_bitmask = 0; + if (revents & EPOLLIN) { + ev_bitmask |= CURL_CSELECT_IN; + } + if (revents & EPOLLOUT) { + ev_bitmask |= CURL_CSELECT_OUT; + } + if (revents & (EPOLLERR | EPOLLHUP)) { + ev_bitmask |= CURL_CSELECT_ERR; + } + + curl_multi_socket_action(self->curl_multi_, fd, ev_bitmask, + &self->curl_running_handles_); + self->DrainCurlMessages(); + + return 0; +} + +int Updater::TimerCallback(CURLM*, long timeout_ms, void* userp) { + auto* self = static_cast(userp); + + if (timeout_ms < 0) { + if (self->curl_timer_source_ != nullptr) { + sd_event_source_set_enabled(self->curl_timer_source_, SD_EVENT_OFF); } - job->worker = std::async(std::launch::async, - [this, job] { return RepackRepoData(job); }); - job->dl_result = DownloadResult::OK; + return 0; } + uint64_t usec_now = 0; + sd_event_now(self->event_, CLOCK_MONOTONIC, &usec_now); + const uint64_t deadline = usec_now + static_cast(timeout_ms) * 1000; + + if (self->curl_timer_source_ == nullptr) { + sd_event_add_time(self->event_, &self->curl_timer_source_, CLOCK_MONOTONIC, + deadline, 0, &Updater::OnTimerFired, self); + } else { + sd_event_source_set_time(self->curl_timer_source_, deadline); + } + sd_event_source_set_enabled(self->curl_timer_source_, SD_EVENT_ONESHOT); + return 0; } -int Updater::Update(const std::string& alpm_config_file, bool force) { - int r, ret = 0; +int Updater::OnTimerFired(sd_event_source*, uint64_t, void* userdata) { + auto* self = static_cast(userdata); + curl_multi_socket_action(self->curl_multi_, CURL_SOCKET_TIMEOUT, 0, + &self->curl_running_handles_); + self->DrainCurlMessages(); + return 0; +} +int Updater::OnRepackEventFd(sd_event_source*, int fd, uint32_t, + void* userdata) { + auto* self = static_cast(userdata); + uint64_t count; + // Just draining the counter to clear readability; the count itself + // doesn't matter since ReapFinishedRepacks() scans every job. + (void)!read(fd, &count, sizeof(count)); + self->ReapFinishedRepacks(); + return 0; +} + +void Updater::EnsureCurlEventSources(sd_event* event) { + if (event_ != nullptr) { + return; + } + event_ = event; + + curl_multi_setopt(curl_multi_, CURLMOPT_SOCKETFUNCTION, + &Updater::SocketCallback); + curl_multi_setopt(curl_multi_, CURLMOPT_SOCKETDATA, this); + curl_multi_setopt(curl_multi_, CURLMOPT_TIMERFUNCTION, + &Updater::TimerCallback); + curl_multi_setopt(curl_multi_, CURLMOPT_TIMERDATA, this); + + repack_eventfd_ = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC); + sd_event_add_io(event_, &repack_eventfd_source_, repack_eventfd_, EPOLLIN, + &Updater::OnRepackEventFd, this); +} + +void Updater::Update(sd_event* event, const std::string& alpm_config_file, + bool force, std::function on_done) { AlpmConfig alpm_config; - ret = AlpmConfig::LoadFromFile(alpm_config_file.c_str(), &alpm_config); - if (ret < 0) { - return 1; + if (AlpmConfig::LoadFromFile(alpm_config_file.c_str(), &alpm_config) < 0) { + on_done(1); + return; } if (alpm_config.repos.empty()) { std::cerr << std::format("error: no repos found in {}\n", alpm_config_file); - return 1; + on_done(1); + return; } if (access(cachedir_.c_str(), W_OK)) { std::cerr << std::format("error: unable to write to {}: {}\n", cachedir_, strerror(errno)); - return 1; + on_done(1); + return; } std::cout << std::format(":: Updating {} repos...\n", @@ -485,84 +621,65 @@ int Updater::Update(const std::string& alpm_config_file, bool force) { // ensure all our DBs are 0644 umask(0022); - auto& repos = alpm_config.repos; - std::vector repo_names; - repo_names.reserve(repos.size()); - for (const auto& repo : repos) { + repo_names.reserve(alpm_config.repos.size()); + for (const auto& repo : alpm_config.repos) { repo_names.push_back(repo.name); } progress_ = std::make_unique(std::move(repo_names)); - // One DownloadJob per repo, holding all state for this update run. - // References into `repos` stay valid: AlpmConfig::LoadFromFile already - // populated it and nothing resizes it afterwards. - std::vector jobs; - jobs.reserve(repos.size()); - for (const auto& repo : repos) { - jobs.emplace_back(repo); + auto state = std::make_shared(); + state->remaining = static_cast(alpm_config.repos.size()); + for (const auto& repo : alpm_config.repos) { + state->known_repos.insert(repo.name); } - // prime the handle by adding a URL from each repo - for (size_t i = 0; i < jobs.size(); ++i) { - DownloadJob& job = jobs[i]; + EnsureCurlEventSources(event); + + size_t i = 0; + for (const auto& repo : alpm_config.repos) { + jobs_.emplace_back(repo); + DownloadJob& job = jobs_.back(); job.arch = alpm_config.architecture; job.force = force; job.progress = progress_.get(); - job.progress_index = i; - r = DownloadQueueRequest(curl_multi_, &job); - if (r != 0) { - ret = r; - } - } + job.progress_index = i++; + job.on_done = [this, state, on_done](const Repo&, DownloadResult result) { + if (result == DownloadResult::ERROR) { + state->ret = 1; + } + if (--state->remaining == 0) { + progress_->Finish(); + TidyCacheDir(state->known_repos); + if (!Database::WriteDatabaseVersion(cachedir_)) { + std::cerr << "warning: failed to write database version marker\n"; + } + on_done(state->ret); + } + }; - DownloadWaitLoop(curl_multi_); - - // remove handles, check for errors - for (auto& job : jobs) { - curl_multi_remove_handle(curl_multi_, job.curl); - curl_easy_cleanup(job.curl); - - switch (job.dl_result) { - case DownloadResult::OK: - case DownloadResult::UPTODATE: - break; - case DownloadResult::ERROR: - ret = 1; - break; - default: - fprintf(stderr, "BUG: unhandled job->dl_result=%d\n", - static_cast(job.dl_result)); - break; + if (DownloadQueueRequest(&job) != 0) { + job.dl_result = DownloadResult::ERROR; + FinishJob(&job); } } - - if (WaitForRepacking(&jobs, !progress_->IsInteractive()) != 0) { - ret = 1; - } - - progress_->Finish(); - - std::set known_repos; - for (const auto& repo : alpm_config.repos) { - known_repos.insert(repo.name); - } - - TidyCacheDir(known_repos); - - if (!Database::WriteDatabaseVersion(cachedir_)) { - std::cerr << "warning: failed to write database version marker\n"; - } - - return ret; } -Updater::Updater(std::string cachedir) : cachedir_(cachedir) { +Updater::Updater(std::string cachedir) : cachedir_(std::move(cachedir)) { curl_global_init(CURL_GLOBAL_ALL); curl_multi_ = curl_multi_init(); } Updater::~Updater() { + if (curl_timer_source_ != nullptr) { + sd_event_source_unref(curl_timer_source_); + } + if (repack_eventfd_source_ != nullptr) { + sd_event_source_unref(repack_eventfd_source_); + } + if (repack_eventfd_ >= 0) { + close(repack_eventfd_); + } curl_multi_cleanup(curl_multi_); curl_global_cleanup(); } diff --git a/src/update.hh b/src/update.hh index 31ad78a..07eaa65 100644 --- a/src/update.hh +++ b/src/update.hh @@ -1,10 +1,13 @@ #pragma once #include +#include #include #include +#include #include +#include #include #include #include @@ -22,20 +25,25 @@ enum class DownloadResult { ERROR, }; -// Runtime state for downloading and repacking one repo, scoped to a single -// Updater::Update() run: the curl handle, destination tmpfile, retry cursor -// into repo.servers, and the async repack worker. +// Runtime state for downloading and repacking one repo: the curl handle, +// destination tmpfile, retry cursor into repo.servers, and the async repack +// worker. Lives in an Updater's job list for as long as its download and (if +// triggered) repack are in flight. struct DownloadJob { - explicit DownloadJob(const Repo& repo) : repo(repo) {} + explicit DownloadJob(Repo repo) : repo(std::move(repo)) {} ~DownloadJob(); + // Never copied or moved: Updater stores these in a std::list so curl + // (CURLOPT_PRIVATE) and in-flight std::async repack workers can hold a + // stable DownloadJob* for the job's whole lifetime. DownloadJob(const DownloadJob&) = delete; DownloadJob& operator=(const DownloadJob&) = delete; - - DownloadJob(DownloadJob&&) = default; + DownloadJob(DownloadJob&&) = delete; DownloadJob& operator=(DownloadJob&&) = delete; - const Repo& repo; + // Owned copy (not a reference into the caller's config) so the job's + // lifetime never depends on the caller keeping an AlpmConfig alive. + Repo repo; std::string arch; // force update repos @@ -48,7 +56,7 @@ struct DownloadJob { // iterator to currently in-use server std::vector::const_iterator server_iter; // error buffer - char errmsg[CURL_ERROR_SIZE]; + char errmsg[CURL_ERROR_SIZE] = {}; // numeric err for determining success DownloadResult dl_result = DownloadResult::UNKNOWN; // start time for download @@ -64,27 +72,89 @@ struct DownloadJob { struct { int fd = -1; - off_t size; + off_t size = 0; } tmpfile; + + // Invoked exactly once, when this job's outcome is final: a download + // failure that exhausted every mirror, an up-to-date result, or a + // completed (successful or failed) repack. + std::function on_done; }; +// Downloads pacman `.files` repo databases and repacks them into the +// pkgfile PFDB cache. +// +// Updater never runs an event loop itself: Update() only registers curl +// transfers and repack completions on an sd_event the caller owns, and +// returns immediately. The caller is responsible for pumping that loop +// (e.g. via sd_event_loop()) until the supplied completion callback fires. +// An Updater instance may only ever be driven by one sd_event over its +// lifetime -- the first event passed to Update() is the one it binds to. class Updater { public: - Updater(std::string cachedir); + explicit Updater(std::string cachedir); ~Updater(); - int Update(const std::string& alpm_config_file, bool force); + Updater(const Updater&) = delete; + Updater& operator=(const Updater&) = delete; + + // Loads `alpm_config_file`, downloads+repacks every repo it declares, and + // invokes `on_done` with the process exit code the old synchronous + // Update() used to return, once every repo has resolved. + void Update(sd_event* event, const std::string& alpm_config_file, bool force, + std::function on_done); private: - int DownloadQueueRequest(CURLM* multi, DownloadJob* job); - void DownloadWaitLoop(CURLM* multi); - int DownloadCheckComplete(CURLM* multi, int remaining); + int DownloadQueueRequest(DownloadJob* job); bool RepackRepoData(const DownloadJob* job); void TidyCacheDir(const std::set& known_repos); + // Removes `job` from the multi handle and frees its curl easy handle, if + // any. Safe to call more than once. + void CleanupCurl(DownloadJob* job); + // Finalizes `job`: cleans up curl, invokes its on_done callback, and + // erases it from jobs_. `job` must not be touched afterward. + void FinishJob(DownloadJob* job); + + void HandleDownloadComplete(DownloadJob* job, CURLMsg* msg); + void DrainCurlMessages(); + void StartRepack(DownloadJob* job); + void ReapFinishedRepacks(); + + // Binds curl's multi handle to `event` via the socket-action interface, + // and creates the repack-completion eventfd source. A no-op after the + // first call. + void EnsureCurlEventSources(sd_event* event); + + static int SocketCallback(CURL* easy, curl_socket_t s, int action, + void* userp, void* socketp); + static int TimerCallback(CURLM* multi, long timeout_ms, void* userp); + static int OnSocketReady(sd_event_source* source, int fd, uint32_t revents, + void* userdata); + static int OnTimerFired(sd_event_source* source, uint64_t usec, + void* userdata); + static int OnRepackEventFd(sd_event_source* source, int fd, uint32_t revents, + void* userdata); + std::string cachedir_; CURLM* curl_multi_; std::unique_ptr progress_; + + // The sd_event this Updater is bound to; see EnsureCurlEventSources(). + sd_event* event_ = nullptr; + sd_event_source* curl_timer_source_ = nullptr; + // Running-handle count as of the most recent curl_multi_socket_action() + // call; used only for the "N remaining" progress print, mirroring what + // curl_multi_perform()'s out-param gave the old blocking loop. + int curl_running_handles_ = 0; + + // Signalled by repack worker threads (via std::async) so the event loop + // can reap finished futures without blocking on them. + int repack_eventfd_ = -1; + sd_event_source* repack_eventfd_source_ = nullptr; + + // Stable storage: see DownloadJob's move/copy note above. + std::list jobs_; }; } // namespace pkgfile