# Acts as a ceiling on the computed polling rate.
poll-max = 1h
+# The [state] section controls hash table and crawl checkpoint persistence.
+
+[state]
+
+ # Enable persistent state (hash table and crawl checkpoints).
+ # When set to no, bees starts fresh every run with an in-memory
+ # hash table.
+ persistent = yes
+
+# The [state.hash] section controls hash table file options.
+
+[state.hash]
+
+ # Hash table size in bytes. Larger tables store more hashes and
+ # find more duplicates. The default uses the smaller of RAM/16
+ # and filesystem-capacity/16384, rounded up to the next 128 KiB
+ # hash-table extent boundary.
+ #
+ # Size value; accepts ${RAM_BYTES}, ${FS_BYTES}, and
+ # other substitutions (see docs/config-format.md).
+ size = min(${RAM_BYTES} / 16, ${FS_BYTES} / 16384)
+
+ # Create beeshash.dat if it does not exist.
+ # Set to no to require the file to be present at startup.
+ create = yes
+
+ # Resize the on-disk hash table to state.hash.size on startup.
+ # Uses reverse-LRU rebuild to preserve the most-recently-used entries.
+ # When no, a size mismatch is logged as a warning and the existing
+ # file size is used instead.
+ resize = no
+
+ # Time budget for writing the full hash table to disk during
+ # normal operation. Limited by writeback-rate-max.
+ writeback-time = 2h
+
+ # Maximum hash table writeback rate in bytes per second. The
+ # effective rate is min(size / writeback-time, writeback-rate-max).
+ writeback-rate-max = 128K
+
+ # fsync after each hash table extent write. Provides a small
+ # increase in durability at the cost of frequent committed writes.
+ writeback-fsync = no
+
+ # Evict hash table pages from the VFS page cache after each write.
+ # Avoids flooding the page cache with hash table data.
+ writeback-unreadahead = no
+
+ # fsync the hash table file when bees exits.
+ close-fsync = no
+
+# The [state.point] section controls crawl checkpoint options.
+
+[state.point]
+
+ # Interval in seconds between crawl checkpoint writes.
+ interval = 900s
+
+ # Defer crawl checkpoint writes until the hash table writeback has
+ # flushed all extents to disk that were modified since the checkpoint update.
+ # Prevents beespoint.ini from advancing past beeshash.dat after a
+ # crash, which would cause missed deduplication on restart.
+ defer = no
+
# The [scan.subvol] section controls legacy subvol crawlers (scan modes 0-3).
[scan.subvol]
# Subvol scans inactive by default.
active = no
- # Subvol scan mode: 0=lockstep, 1=independent, 2=sequential, 3=recent (legacy -m values 0-3).
+ # Subvol scan mode: 0=lockstep, 1=independent, 2=sequential,
+ # 3=recent (legacy -m values 0-3).
mode = 3
# Skip read-only subvols entirely.
#include "crucible/error.h"
#include "crucible/fs.h"
+#include "crucible/limits.h"
#include "crucible/path.h"
#include "crucible/string.h"
#include "crucible/uuid.h"
+#include <cmath>
#include <sstream>
+#include <sys/statfs.h>
#define THROW_CONFIG_ERROR(__key, __value, __message) do { \
ostringstream __oss; \
}
insert_map_unique(m_subst_map, "LABEL", fslabel);
+ BEESTRACE("Getting RAM_BYTES from sysconf");
+ const auto phys_pages = sysconf(_SC_PHYS_PAGES);
+ const auto page_size = sysconf(_SC_PAGE_SIZE);
+ THROW_CHECK2(runtime_error, phys_pages, page_size, phys_pages > 0 && page_size > 0);
+ const auto ram_bytes = static_cast<uint64_t>(phys_pages) * static_cast<uint64_t>(page_size);
+ insert_map_unique(m_subst_map, "RAM_BYTES", to_string(ram_bytes));
+
+ BEESTRACE("Getting FS_BYTES from " << name_fd(m_root_fd));
+ struct statfs fs_stat = {};
+ DIE_IF_NON_ZERO(fstatfs(m_root_fd, &fs_stat));
+ const auto fs_bytes = ranged_cast<uint64_t>(fs_stat.f_blocks) * ranged_cast<uint64_t>(fs_stat.f_frsize);
+ insert_map_unique(m_subst_map, "FS_BYTES", to_string(fs_bytes));
+
for (const auto &i : m_subst_map) {
BEESLOGDEBUG("subst map: '" << i.first << "' = '" << i.second << "'");
}
THROW_ERROR(invalid_argument, "'" << str << "' not recognized as bool value");
}
-/// Parse a single size token (digits with optional K/M/G/T/P/E suffix).
-static
-uint64_t
-parse_size_token(const string &tok, const string &context)
-{
- if (tok.empty()) {
- THROW_ERROR(invalid_argument, "empty size value in '" << context << "'");
+/// Recursive descent parser for size expressions.
+/// Grammar (standard arithmetic precedence):
+/// expr := add_expr
+/// add_expr := mul_expr ( ('+' | '-') mul_expr )*
+/// mul_expr := atom ( ('*' | '/') atom )*
+/// atom := NUMBER SUFFIX?
+/// | '(' expr ')'
+/// | 'min' '(' expr ',' expr ')'
+/// | 'max' '(' expr ',' expr ')'
+/// | 'max' -- numeric_limits<uint64_t>::max()
+/// SUFFIX := K|M|G|T|P|E (powers of 1024) | '%' (divide by 100)
+///
+/// All arithmetic uses double; the final result is cast to uint64_t.
+/// 'max' alone evaluates to infinity, which maps to uint64_t max on exit.
+namespace {
+struct SizeExprParser {
+ const string &input;
+ size_t pos;
+
+ explicit SizeExprParser(const string &s) : input(s), pos(0) {}
+
+ void skip_ws() {
+ while (pos < input.size() && isspace(static_cast<unsigned char>(input[pos])))
+ ++pos;
}
- const char suffix = tok.back();
- uint64_t multiplier = 1;
- string digits = tok;
- switch (toupper(static_cast<unsigned char>(suffix))) {
- case 'K': multiplier = 1ULL << 10; break;
- case 'M': multiplier = 1ULL << 20; break;
- case 'G': multiplier = 1ULL << 30; break;
- case 'T': multiplier = 1ULL << 40; break;
- case 'P': multiplier = 1ULL << 50; break;
- case 'E': multiplier = 1ULL << 60; break;
- default:
- THROW_CHECK1(invalid_argument, suffix, isdigit(suffix));
- break;
+ bool at_end() { skip_ws(); return pos >= input.size(); }
+ char peek() { skip_ws(); return pos < input.size() ? input[pos] : '\0'; }
+
+ bool try_consume(char c) {
+ if (peek() == c) { ++pos; return true; }
+ return false;
}
- if (multiplier != 1) {
- digits = tok.substr(0, tok.size() - 1);
+
+ double parse_expr() { return parse_add(); }
+
+ double parse_add() {
+ double v = parse_mul();
+ while (true) {
+ if (try_consume('+')) {
+ v += parse_mul();
+ } else if (peek() == '-') {
+ ++pos;
+ v -= parse_mul();
+ } else {
+ break;
+ }
+ }
+ return v;
}
- return stoull(digits) * multiplier;
-}
+ double parse_mul() {
+ double v = parse_atom();
+ while (true) {
+ if (try_consume('*')) {
+ v *= parse_atom();
+ } else if (try_consume('/')) {
+ const double d = parse_atom();
+ THROW_CHECK1(invalid_argument, d, d != 0.0);
+ v /= d;
+ } else {
+ break;
+ }
+ }
+ return v;
+ }
+
+ double parse_atom() {
+ skip_ws();
+ if (pos >= input.size()) {
+ THROW_ERROR(invalid_argument, "unexpected end of size expression '" << input << "'");
+ }
+
+ // Parenthesized sub-expression
+ if (input[pos] == '(') {
+ ++pos;
+ const double v = parse_expr();
+ skip_ws();
+ if (pos >= input.size() || input[pos] != ')') {
+ THROW_ERROR(invalid_argument, "missing ')' in size expression '" << input << "'");
+ }
+ ++pos;
+ return v;
+ }
+
+ // Identifier: 'min'/'max' functions or 'max' sentinel
+ if (isalpha(static_cast<unsigned char>(input[pos]))) {
+ const size_t id_start = pos;
+ while (pos < input.size() && isalpha(static_cast<unsigned char>(input[pos])))
+ ++pos;
+ const string ident = input.substr(id_start, pos - id_start);
+
+ skip_ws();
+ if (pos < input.size() && input[pos] == '(') {
+ ++pos;
+ const double a = parse_expr();
+ skip_ws();
+ if (pos >= input.size() || input[pos] != ',') {
+ THROW_ERROR(invalid_argument, "expected ',' in '" << ident << "()' in '" << input << "'");
+ }
+ ++pos;
+ const double b = parse_expr();
+ skip_ws();
+ if (pos >= input.size() || input[pos] != ')') {
+ THROW_ERROR(invalid_argument, "missing ')' after '" << ident << "()' in '" << input << "'");
+ }
+ ++pos;
+ if (ident == "min") return min(a, b);
+ if (ident == "max") return max(a, b);
+ THROW_ERROR(invalid_argument, "unknown function '" << ident << "' in '" << input << "'");
+ }
+
+ // 'max' as standalone keyword = uint64_t max
+ if (ident == "max") {
+ return numeric_limits<double>::infinity();
+ }
+
+ THROW_ERROR(invalid_argument, "unexpected identifier '" << ident << "' in size expression '" << input << "'");
+ }
+
+ // Numeric literal with optional size or percentage suffix
+ if (isdigit(static_cast<unsigned char>(input[pos])) || input[pos] == '.') {
+ const size_t num_start = pos;
+ while (pos < input.size() &&
+ (isdigit(static_cast<unsigned char>(input[pos])) || input[pos] == '.'))
+ ++pos;
+ double value = stod(input.substr(num_start, pos - num_start));
+
+ if (pos < input.size()) {
+ const char raw = input[pos];
+ const char suffix = static_cast<char>(toupper(static_cast<unsigned char>(raw)));
+ switch (suffix) {
+ case 'K': value *= double(1ULL << 10); ++pos; break;
+ case 'M': value *= double(1ULL << 20); ++pos; break;
+ case 'G': value *= double(1ULL << 30); ++pos; break;
+ case 'T': value *= double(1ULL << 40); ++pos; break;
+ case 'P': value *= double(1ULL << 50); ++pos; break;
+ case 'E': value *= double(1ULL << 60); ++pos; break;
+ case '%': value /= 100.0; ++pos; break;
+ default: break;
+ }
+ }
+ return value;
+ }
+
+ THROW_ERROR(invalid_argument, "unexpected character '" << input[pos] << "' in size expression '" << input << "'");
+ }
+};
+} // anonymous namespace
uint64_t
bees_parse_size(const string &s)
{
BEESTRACE("parsing size '" << s << "'");
- if (s == "max") {
- return numeric_limits<uint64_t>::max();
- }
- // Find a '+' or '-' operator after the base value (e.g. "512K + 1",
- // "128K - 4K"). The operand may also carry a scale suffix.
- string base = s;
- char op = 0;
- string operand;
- for (auto c : "+-") {
- const auto pos = s.find(c);
- if (pos != string::npos) {
- op = c;
- base = s.substr(0, pos);
- operand = s.substr(pos + 1);
- break;
- }
+ SizeExprParser parser(s);
+ if (parser.at_end()) {
+ THROW_ERROR(invalid_argument, "empty size expression");
}
- // Trim whitespace around both tokens.
- while (!base.empty() && isspace(static_cast<unsigned char>(base.back()))) base.pop_back();
- while (!operand.empty() && isspace(static_cast<unsigned char>(operand.front()))) operand.erase(operand.begin());
-
- const uint64_t base_val = parse_size_token(base, s);
+ const double result = parser.parse_expr();
- if (!op) {
- return base_val;
+ if (!parser.at_end()) {
+ THROW_ERROR(invalid_argument, "unexpected trailing characters in size expression '" << s << "'");
}
- const uint64_t operand_val = parse_size_token(operand, s);
- if (op == '+') {
- return base_val + operand_val;
+ if (isinf(result)) {
+ return numeric_limits<uint64_t>::max();
}
- // op == '-'
- THROW_CHECK2(invalid_argument, base_val, operand_val, base_val >= operand_val);
- return base_val - operand_val;
+
+ THROW_CHECK1(invalid_argument, result, result >= 0.0);
+ return static_cast<uint64_t>(result);
}
uint64_t
BeesContext::hash_table()
{
unique_lock<mutex> lock(m_stop_mutex);
- if (!m_hash_table) {
- m_hash_table = make_shared<BeesHashTable>(shared_from_this(), "beeshash.dat");
- }
+ THROW_CHECK0(runtime_error, m_hash_table);
return m_hash_table;
}
+void
+BeesContext::init_hash_table(const string &filename, off_t size)
+{
+ unique_lock<mutex> lock(m_stop_mutex);
+ m_hash_table = make_shared<BeesHashTable>(shared_from_this(), filename, size, BeesHashTable::NoThreadsTag{});
+}
+
+void
+BeesContext::init_hash_table_memory(off_t size)
+{
+ unique_lock<mutex> lock(m_stop_mutex);
+ m_hash_table = make_shared<BeesHashTable>(shared_from_this(), size, BeesHashTable::NoFileTag{});
+}
+
+void
+BeesContext::start_hash_writeback()
+{
+ hash_table()->start_threads();
+}
+
+void
+BeesContext::flush_hash_table()
+{
+ hash_table()->flush();
+}
+
+void
+BeesContext::start_crawl()
+{
+ roots()->start();
+}
+
+void
+BeesContext::stop_crawl()
+{
+ roots()->stop_request();
+ roots()->stop_wait();
+}
+
void
BeesContext::set_root_path(const string& path)
{
auto lock = lock_extent_by_index(extent_index);
bool wrote_extent = false;
+ if (!m_persistent) {
+ // Memory-only mode: mark clean without any file I/O.
+ m_extent_metadata.at(extent_index).m_dirty = false;
+ return true;
+ }
+
catch_all([&]() {
uint8_t *const dirty_extent = m_extent_ptr[extent_index].p_byte;
uint8_t *const dirty_extent_end = m_extent_ptr[extent_index + 1].p_byte;
const size_t dirty_extent_offset = dirty_extent - m_byte_ptr;
+ const size_t dirty_extent_size = dirty_extent_end - dirty_extent;
THROW_CHECK1(out_of_range, dirty_extent, dirty_extent >= m_byte_ptr);
THROW_CHECK1(out_of_range, dirty_extent_end, dirty_extent_end <= m_byte_ptr_end);
THROW_CHECK2(out_of_range, dirty_extent_end, dirty_extent, dirty_extent_end - dirty_extent == BLOCK_SIZE_HASHTAB_EXTENT);
- BEESTOOLONG("pwrite(fd " << m_fd << " '" << name_fd(m_fd)<< "', length " << to_hex(dirty_extent_end - dirty_extent) << ", offset " << to_hex(dirty_extent - m_byte_ptr) << ")");
+ BEESTOOLONG("pwrite(fd " << m_fd << " '" << name_fd(m_fd)<< "', length " << to_hex(dirty_extent_size) << ", offset " << to_hex(dirty_extent_offset) << ")");
// Copy the extent because we might be stuck writing for a while
ByteVector extent_copy(dirty_extent, dirty_extent_end);
// Release the lock
lock.unlock();
- // Write the extent (or not)
+ // Write the extent
pwrite_or_die(m_fd, extent_copy, dirty_extent_offset);
BEESCOUNT(hash_extent_out);
- // Nope, this causes a _dramatic_ loss of performance.
- // const size_t dirty_extent_size = dirty_extent_end - dirty_extent;
- // bees_unreadahead(m_fd, dirty_extent_offset, dirty_extent_size);
+ if (m_writeback_unreadahead) {
+ bees_unreadahead(m_fd, dirty_extent_offset, dirty_extent_size);
+ }
+ if (m_writeback_fsync) {
+ bees_fsync(m_fd);
+ }
// Mark extent clean if write was successful
lock.lock();
// Skip the clean ones
auto lock = lock_extent_by_index(extent_index);
if (!m_extent_metadata.at(extent_index).m_dirty) {
+ ++m_writeback_extent_count;
continue;
}
lock.unlock();
if (flush_dirty_extent(extent_index)) {
++wrote_extents;
+ ++m_writeback_extent_count;
if (slowly) {
if (m_stop_requested) {
slowly = false;
return;
}
+ if (!m_persistent) {
+ // Memory-only mode: extent is zero-initialised from MAP_ANONYMOUS.
+ m_extent_metadata.at(extent_index).m_missing = false;
+ return;
+ }
+
// OK we have to read this extent
BEESNOTE("fetching hash extent #" << extent_index << " of " << m_extents << " extents");
BEESTRACE("Fetching hash extent #" << extent_index << " of " << m_extents << " extents");
// If that doesn't work, try to make a new one
if (!new_fd) {
+ if (!m_create) {
+ THROW_ERRNO("hash table '" << m_filename << "' does not exist and state.hash.create = no");
+ }
string tmp_filename = m_filename + ".tmp";
BEESNOTE("creating new hash table '" << tmp_filename << "'");
BEESLOGINFO("Creating new hash table '" << tmp_filename << "'");
m_fd = new_fd;
}
-BeesHashTable::BeesHashTable(shared_ptr<BeesContext> ctx, string filename, off_t size) :
- m_ctx(ctx),
- m_size(0),
- m_void_ptr(nullptr),
- m_void_ptr_end(nullptr),
- m_buckets(0),
- m_cells(0),
- m_writeback_thread("hash_writeback"),
- m_prefetch_thread("hash_prefetch"),
- m_flush_rate_limit(BEES_FLUSH_RATE),
- m_stats_file(m_ctx->home_fd(), "beesstats.txt")
+void
+BeesHashTable::init(string filename, off_t size)
{
// Sanity checks to protect the implementation from its weaknesses
THROW_CHECK2(invalid_argument, BLOCK_SIZE_HASHTAB_BUCKET, BLOCK_SIZE_HASHTAB_EXTENT, (BLOCK_SIZE_HASHTAB_EXTENT % BLOCK_SIZE_HASHTAB_BUCKET) == 0);
THROW_CHECK2(runtime_error, sizeof(Extent::p_byte), BLOCK_SIZE_HASHTAB_EXTENT, BLOCK_SIZE_HASHTAB_EXTENT == sizeof(Extent::p_byte));
m_filename = filename;
+ const uint64_t desired_size = static_cast<uint64_t>(size);
m_size = size;
- open_file();
+ if (m_persistent) {
+ open_file();
+ if (m_resize && m_size != desired_size) {
+ resize_file(static_cast<off_t>(desired_size));
+ } else if (m_size != desired_size) {
+ BEESLOGWARN("hash table on-disk size " << pretty(m_size) << " differs from configured state.hash.size " << pretty(desired_size) << " (use state.hash.resize = yes to resize)");
+ }
+ }
// Now we know size we can compute stuff
}
m_extent_metadata.resize(m_extents);
+}
+void
+BeesHashTable::start_threads()
+{
m_writeback_thread.exec([&]() {
writeback_loop();
});
m_prefetch_thread.exec([&]() {
prefetch_loop();
});
+}
+
+void
+BeesHashTable::flush()
+{
+ flush_dirty_extents(false);
+}
+
+BeesHashTable::BeesHashTable(shared_ptr<BeesContext> ctx, string filename, off_t size) :
+ m_ctx(ctx),
+ m_size(0),
+ m_void_ptr(nullptr),
+ m_void_ptr_end(nullptr),
+ m_buckets(0),
+ m_cells(0),
+ m_writeback_thread("hash_writeback"),
+ m_prefetch_thread("hash_prefetch"),
+ m_flush_rate_limit(BEES_FLUSH_RATE),
+ m_stats_file(m_ctx->home_fd(), "beesstats.txt")
+{
+ init(filename, size);
+ start_threads();
// Blacklist might fail if the hash table is not stored on a btrfs,
// or if it's on a _different_ btrfs
});
}
+BeesHashTable::BeesHashTable(shared_ptr<BeesContext> ctx, string filename, off_t size, NoThreadsTag) :
+ m_ctx(ctx),
+ m_size(0),
+ m_void_ptr(nullptr),
+ m_void_ptr_end(nullptr),
+ m_buckets(0),
+ m_cells(0),
+ m_writeback_thread("hash_writeback"),
+ m_prefetch_thread("hash_prefetch"),
+ m_flush_rate_limit(BEES_FLUSH_RATE),
+ m_stats_file(m_ctx->home_fd(), "beesstats.txt")
+{
+ init(filename, size);
+}
+
+BeesHashTable::BeesHashTable(shared_ptr<BeesContext> ctx, off_t size, NoFileTag) :
+ m_ctx(ctx),
+ m_size(0),
+ m_void_ptr(nullptr),
+ m_void_ptr_end(nullptr),
+ m_buckets(0),
+ m_cells(0),
+ m_writeback_thread("hash_writeback"),
+ m_prefetch_thread("hash_prefetch"),
+ m_flush_rate_limit(BEES_FLUSH_RATE),
+ m_stats_file(m_ctx->home_fd(), "beesstats.txt")
+{
+ m_persistent = false;
+ init("", size);
+}
+
+void
+BeesHashTable::set_flush_rate(double bps)
+{
+ m_flush_rate_limit.rate(bps);
+}
+
+void
+BeesHashTable::resize_file(off_t new_size)
+{
+ THROW_CHECK1(runtime_error, m_persistent, m_persistent);
+ if (m_size == static_cast<uint64_t>(new_size)) {
+ return;
+ }
+
+ const off_t old_size = static_cast<off_t>(m_size);
+ const uint64_t old_extents = static_cast<uint64_t>(old_size) / BLOCK_SIZE_HASHTAB_EXTENT;
+ const uint64_t new_buckets = static_cast<uint64_t>(new_size) / BLOCK_SIZE_HASHTAB_BUCKET;
+
+ BEESLOGINFO("Resizing hash table from " << pretty(old_size) << " (" << old_extents << " extents) to " << pretty(new_size));
+
+ // Allocate new table in RAM, zero-initialized.
+ vector<uint8_t> new_table(new_size, 0);
+ Bucket *const new_bucket_ptr = reinterpret_cast<Bucket *>(new_table.data());
+
+ // Insert one (hash, addr) cell into new_table using push-front LRU.
+ auto insert_cell = [&](HashType hash, AddrType addr) {
+ if (!addr) return;
+ Bucket &bucket = new_bucket_ptr[hash % new_buckets];
+ // If already present, don't duplicate.
+ for (uint64_t i = 0; i < c_cells_per_bucket; ++i) {
+ if (bucket.p_cells[i].e_hash == hash && bucket.p_cells[i].e_addr == addr) {
+ return;
+ }
+ }
+ // Shift existing cells right, insert at front (evict the last).
+ for (uint64_t i = c_cells_per_bucket - 1; i > 0; --i) {
+ bucket.p_cells[i] = bucket.p_cells[i - 1];
+ }
+ bucket.p_cells[0] = Cell(hash, addr);
+ };
+
+ // Read old extents in reverse order: oldest (lowest LRU priority) first,
+ // newest (highest LRU priority) last, so newest entries win on collision.
+ ByteVector extent_buf(BLOCK_SIZE_HASHTAB_EXTENT);
+ for (uint64_t pass = 0; pass < old_extents; ++pass) {
+ const uint64_t ei = old_extents - 1 - pass;
+ BEESNOTE("resizing hash table: reading extent " << (pass + 1) << " of " << old_extents);
+ const off_t offset = static_cast<off_t>(ei) * BLOCK_SIZE_HASHTAB_EXTENT;
+ pread_or_die(m_fd, extent_buf, offset);
+
+ const uint64_t buckets_per_extent = BLOCK_SIZE_HASHTAB_EXTENT / BLOCK_SIZE_HASHTAB_BUCKET;
+ const Bucket *old_buckets = reinterpret_cast<const Bucket *>(extent_buf.data());
+ for (uint64_t b = 0; b < buckets_per_extent; ++b) {
+ for (uint64_t c = 0; c < c_cells_per_bucket; ++c) {
+ insert_cell(old_buckets[b].p_cells[c].e_hash, old_buckets[b].p_cells[c].e_addr);
+ }
+ }
+ }
+
+ // Write the new table to a temp file, fsync, rename over the original.
+ const string tmp_filename = m_filename + ".resize";
+ BEESNOTE("writing resized hash table to '" << tmp_filename << "'");
+ BEESLOGINFO("Writing resized hash table to '" << tmp_filename << "'");
+ unlinkat(m_ctx->home_fd(), tmp_filename.c_str(), 0);
+ Fd tmp_fd = openat_or_die(m_ctx->home_fd(), tmp_filename, FLAGS_CREATE_FILE, 0700);
+ ftruncate_or_die(tmp_fd, new_size);
+ const uint64_t new_extents = static_cast<uint64_t>(new_size) / BLOCK_SIZE_HASHTAB_EXTENT;
+ for (uint64_t ei = 0; ei < new_extents; ++ei) {
+ BEESNOTE("writing resized hash table: extent " << (ei + 1) << " of " << new_extents);
+ const off_t write_offset = static_cast<off_t>(ei) * BLOCK_SIZE_HASHTAB_EXTENT;
+ pwrite_or_die(tmp_fd, new_table.data() + write_offset, BLOCK_SIZE_HASHTAB_EXTENT, write_offset);
+ }
+ bees_fsync(tmp_fd);
+ tmp_fd = Fd();
+ renameat_or_die(m_ctx->home_fd(), tmp_filename, m_ctx->home_fd(), m_filename);
+
+ // Reopen the renamed file and update m_size.
+ m_fd = openat_or_die(m_ctx->home_fd(), m_filename, FLAGS_OPEN_FILE_RW, 0700);
+ m_size = new_size;
+ BEESLOGINFO("Hash table resize complete: " << pretty(old_size) << " -> " << pretty(new_size));
+}
+
BeesHashTable::~BeesHashTable()
{
BEESLOGDEBUG("Destroy BeesHashTable");
BEESLOGDEBUG("Waiting for hash_writeback thread");
m_writeback_thread.join();
+ if (m_persistent && m_close_fsync && m_fd) {
+ BEESNOTE("fsync hash table on close");
+ BEESLOGDEBUG("fsync hash table on close");
+ catch_all([&]() { bees_fsync(m_fd); });
+ }
+
BEESLOGDEBUG("BeesHashTable stopped");
}
BEESLOGINFO("poll-max = " << max_seconds << "s [loop.poll-max]");
}
+void
+BeesRoots::set_checkpoint_interval(double seconds)
+{
+ THROW_CHECK1(invalid_argument, seconds, seconds >= 0);
+ m_checkpoint_interval = seconds;
+ BEESLOGINFO("checkpoint interval = " << m_checkpoint_interval << "s [state.point.interval]");
+}
+
+void
+BeesRoots::set_persistent(bool v)
+{
+ BEESLOGINFO("state persistence = " << (v ? "yes" : "no") << " [state.persistent]");
+ m_roots_persistent = v;
+}
+
+void
+BeesRoots::set_checkpoint_defer(bool v)
+{
+ BEESLOGINFO("state defer = " << (v ? "yes" : "no") << " [state.point.defer]");
+ m_checkpoint_defer = v;
+}
+
uint64_t
BeesRoots::effective_transid_max()
{
bc->set_reporter(reporter);
}
+ // Set up persistent state (hash table and crawl checkpoints)
+ {
+ const bool persistent = bc->get_config().get("state.persistent", bees_parse_bool);
+ // Round up to the next hash table extent boundary (128 KiB).
+ const auto hash_size_raw = static_cast<off_t>(bc->get_config().get("state.hash.size", [&](const string &s) {
+ return bees_parse_size(bc->get_config().subst(s));
+ }));
+ const auto hash_size = (hash_size_raw + BLOCK_SIZE_HASHTAB_EXTENT - 1)
+ / BLOCK_SIZE_HASHTAB_EXTENT * BLOCK_SIZE_HASHTAB_EXTENT;
+
+ if (!persistent) {
+ // Pure in-memory mode: allocate hash table in RAM, skip all file I/O.
+ bc->init_hash_table_memory(hash_size);
+ bc->roots()->set_persistent(false);
+ } else {
+ // File-backed mode: open (or create) beeshash.dat.
+ bc->init_hash_table("beeshash.dat", hash_size);
+ auto ht = bc->hash_table();
+ ht->set_create(bc->get_config().get("state.hash.create", bees_parse_bool));
+ ht->set_resize(bc->get_config().get("state.hash.resize", bees_parse_bool));
+ ht->set_writeback_fsync(bc->get_config().get("state.hash.writeback-fsync", bees_parse_bool));
+ ht->set_writeback_unreadahead(bc->get_config().get("state.hash.writeback-unreadahead", bees_parse_bool));
+ ht->set_close_fsync(bc->get_config().get("state.hash.close-fsync", bees_parse_bool));
+ // Compute effective writeback rate: min(size / writeback-time, writeback-rate-max)
+ const double writeback_time = bc->get_config().get("state.hash.writeback-time", bees_parse_duration);
+ const double writeback_rate_max = static_cast<double>(bc->get_config().get("state.hash.writeback-rate-max", bees_parse_size));
+ const double computed_rate = writeback_time > 0 ? static_cast<double>(hash_size) / writeback_time : writeback_rate_max;
+ ht->set_flush_rate(min(computed_rate, writeback_rate_max));
+ }
+ bc->start_hash_writeback();
+
+ // Crawl checkpoint options
+ bc->roots()->set_checkpoint_interval(
+ bc->get_config().get("state.point.interval", bees_parse_duration));
+ bc->roots()->set_checkpoint_defer(
+ bc->get_config().get("state.point.defer", bees_parse_bool));
+ }
+
// Workaround for the logical-ino-vs-clone kernel bug
MultiLocker::enable_locking(true);
#include "crucible/task.h"
#include <array>
+#include <atomic>
#include <functional>
#include <list>
#include <mutex>
/// Open or create the hash table file at @p filename with the given @p size.
BeesHashTable(shared_ptr<BeesContext> ctx, string filename, off_t size = BLOCK_SIZE_HASHTAB_EXTENT);
+
+ /// Tag type: pass as fourth argument to construct without starting threads.
+ struct NoThreadsTag {};
+ /// Like BeesHashTable(ctx, filename, size) but does not start background threads.
+ /// Call start_threads() later to start writeback and prefetch threads.
+ BeesHashTable(shared_ptr<BeesContext> ctx, string filename, off_t size, NoThreadsTag);
+
~BeesHashTable();
/// Signal background threads to stop.
void stop_request();
/// Block until background threads have stopped.
void stop_wait();
+ /// Start writeback and prefetch background threads (call once after construction).
+ void start_threads();
+ /// Flush all dirty extents to disk synchronously (no rate limiting).
+ void flush();
/// Return all Cells whose hash matches @p hash.
vector<Cell> find_cell(HashType hash);
/// Write the extent at @p extent_index to disk if dirty; return true if written.
bool flush_dirty_extent(uint64_t extent_index);
+ /// Tag type: pass as third argument to construct a memory-only table (no file I/O).
+ struct NoFileTag {};
+ /// Like BeesHashTable(ctx, filename, size, NoThreadsTag) but with no file backing.
+ /// The table lives entirely in RAM; all writeback file options are irrelevant.
+ BeesHashTable(shared_ptr<BeesContext> ctx, off_t size, NoFileTag);
+
+ /// Total number of extents in the hash table.
+ uint64_t extent_count() const { return m_extents; }
+ /// Running total of extents confirmed consistent with on-disk state since startup.
+ /// Incremented for each written extent and for each clean (skipped) extent.
+ uint64_t writeback_extent_count() const {
+ return m_writeback_extent_count.load(memory_order_relaxed);
+ }
+ /// If false, table lives in RAM only; all file options are ignored (call before start_threads()).
+ void set_persistent(bool v) { m_persistent = v; }
+ /// Set writeback rate in bytes per second.
+ void set_flush_rate(double bps);
+ /// If false, throw at startup when beeshash.dat is absent instead of creating it.
+ void set_create(bool v) { m_create = v; }
+ /// If true, rebuild the on-disk table at state.hash.size on startup (reverse-LRU merge).
+ void set_resize(bool v) { m_resize = v; }
+ /// fsync after each extent write.
+ void set_writeback_fsync(bool v) { m_writeback_fsync = v; }
+ /// Discard hash table pages from VFS page cache after each write.
+ void set_writeback_unreadahead(bool v) { m_writeback_unreadahead = v; }
+ /// fsync the hash table on close (at bees termination only).
+ void set_close_fsync(bool v) { m_close_fsync = v; }
+
private:
string m_filename;
Fd m_fd;
condition_variable m_stop_condvar;
bool m_stop_requested = false;
+ // Options from [state.hash.*] config
+ atomic<uint64_t> m_writeback_extent_count{0};
+ bool m_persistent = true;
+ bool m_create = true;
+ bool m_resize = false;
+ bool m_writeback_fsync = false;
+ bool m_writeback_unreadahead = false;
+ bool m_close_fsync = false;
+
/// Per-extent metadata tracked in memory alongside the mmap'd data.
struct ExtentMetaData {
shared_ptr<mutex> m_mutex_ptr; ///< Serializes concurrent access to this extent.
};
vector<ExtentMetaData> m_extent_metadata;
+ void init(string filename, off_t size);
void open_file();
void writeback_loop();
void prefetch_loop();
void fetch_missing_extent_by_index(uint64_t extent_index);
void set_extent_dirty_locked(uint64_t extent_index);
size_t flush_dirty_extents(bool slowly);
+ void resize_file(off_t new_size);
size_t hash_to_extent_index(HashType ht);
unique_lock<mutex> lock_extent_by_hash(HashType ht);
bool m_exit_one_pass = false; ///< Exit after all crawlers complete one pass.
uint64_t m_min_transid_age = 0; ///< Stay this many transids behind transid_max().
uint64_t m_min_transid_span = 0; ///< Minimum transid window before starting a new pass (0=any).
+ double m_checkpoint_interval = BEES_WRITEBACK_INTERVAL; ///< Seconds between checkpoint writes.
+ bool m_roots_persistent = true; ///< If false, skip load/save of beespoint.ini.
+ bool m_checkpoint_defer = false; ///< If true, defer writes until hash writeback catches up.
vector<shared_ptr<BeesScanMode>> m_scanners; ///< Active scan-mode strategy objects.
void set_min_transid_span(uint64_t span);
/// Set poll interval bounds in seconds for the transid poll loop.
void set_poll_bounds(double min_seconds, double max_seconds);
+ /// Set the interval in seconds between crawl checkpoint writes.
+ void set_checkpoint_interval(double seconds);
+ /// Disable beespoint.ini persistence (skip load and save).
+ void set_persistent(bool v);
+ /// Defer checkpoint writes until the hash table writeback has lapped past the snapshot point.
+ void set_checkpoint_defer(bool v);
/// Return the lowest transaction ID seen across all active crawls.
uint64_t transid_min();
/// Return a reference to the parsed configuration.
const BeesConfig &get_config() const;
+ /// Discrete-init API. These are hooks that split
+ /// BeesContext::start() / stop() into pieces to enable future
+ /// extensions like hash table resize and non-persistent hash tables.
+ void init_hash_table(const string &filename, off_t size);
+ void init_hash_table_memory(off_t size);
+ void start_hash_writeback();
+ void flush_hash_table();
+ void load_state();
+ void save_state();
+ void print_state(ostream &os);
+ void start_crawl();
+ void stop_crawl();
+
/// Return the FD to the btrfs root mount point.
Fd root_fd() const { return m_root_fd; }
/// Return the directory fd used for O_TMPFILE creation (from tempfile.dir).
projects / bees / commitdiff