void
BeesRoots::set_checkpoint_interval(double seconds)
{
- THROW_CHECK1(invalid_argument, seconds, seconds >= 0);
+ THROW_CHECK1(invalid_argument, seconds, seconds > 0);
m_checkpoint_interval = seconds;
BEESLOGINFO("checkpoint interval = " << m_checkpoint_interval << "s [state.point.interval]");
}
void
BeesRoots::set_checkpoint_defer(bool v)
{
- BEESLOGINFO("state defer = " << (v ? "yes" : "no") << " [state.point.defer]");
+ BEESLOGINFO("defer checkpoint = " << (v ? "yes" : "no") << " [state.point.defer]");
m_checkpoint_defer = v;
}
+void
+BeesRoots::flush_deferred_checkpoints()
+{
+ flush_pending_checkpoints(true);
+}
+
uint64_t
BeesRoots::effective_transid_max()
{
}
void
-BeesRoots::save_point()
+BeesRoots::snapshot_point()
{
- BEESNOTE("saving beespoint.ini");
- BEESLOGINFO("Saving beespoint.ini");
- BEESTOOLONG("Saving beespoint.ini");
-
- Timer save_time;
-
- unique_lock<mutex> lock(m_mutex);
+ // Must already be holding m_mutex.
if (m_crawl_clean == m_crawl_dirty) {
BEESLOGINFO("Nothing to save");
m_point_ini.remove_section(section);
}
} else {
- BEESLOGWARN("save_point: unrecognised root " << ibcs.m_root << ", skipping");
+ BEESLOGWARN("snapshot_point: unrecognised root " << ibcs.m_root << ", skipping");
}
}
const string ini_string = m_point_ini.write();
const auto crawl_saved = m_crawl_dirty;
- lock.unlock();
- m_point_file.write(ini_string);
+ if (!m_checkpoint_defer) {
+ // Immediate write (current behaviour): write while lock is temporarily released.
+ // We must re-acquire the lock before updating m_crawl_clean.
+ // Use a raw unlock/lock here since snapshot_point() is called with lock held.
+ m_mutex.unlock();
+ m_point_file.write(ini_string);
+ m_mutex.lock();
+ m_crawl_clean = crawl_saved;
+ } else {
+ // Deferred write: enqueue with current writeback counter.
+ const auto wc = m_ctx->hash_table()->writeback_extent_count();
+ m_pending_checkpoints.push_back({ wc, ini_string });
+ m_crawl_clean = crawl_saved;
+ }
+}
- lock.lock();
- m_crawl_clean = crawl_saved;
+void
+BeesRoots::flush_pending_checkpoints(bool force)
+{
+ while (true) {
+ string last_ini;
+ {
+ unique_lock<mutex> lock(m_mutex);
+ if (m_pending_checkpoints.empty()) return;
+
+ const auto extent_count = m_ctx->hash_table()->extent_count();
+ const auto writeback_now = m_ctx->hash_table()->writeback_extent_count();
+
+ // Drain all entries eligible under writeback_now (or all if forced),
+ // keeping only the last one's content to write.
+ while (!m_pending_checkpoints.empty()) {
+ const auto &entry = m_pending_checkpoints.front();
+ if (!force && writeback_now - entry.writeback_count < extent_count) {
+ break;
+ }
+ last_ini = std::move(entry.ini_string);
+ m_pending_checkpoints.pop_front();
+ }
+ } // lock released here
+
+ if (last_ini.empty()) return; // nothing became eligible
+ m_point_file.write(last_ini); // slow I/O outside the lock
+ // loop: more entries may be eligible now
+ }
+}
+
+void
+BeesRoots::save_point()
+{
+ BEESNOTE("saving beespoint.ini");
+ BEESLOGINFO("Saving beespoint.ini");
+ BEESTOOLONG("Saving beespoint.ini");
+
+ if (!m_roots_persistent) {
+ return;
+ }
+
+ Timer save_time;
+
+ m_mutex.lock();
+ snapshot_point(); // enqueue or write immediately (may unlock/relock m_mutex)
+ m_mutex.unlock();
+
+ if (m_checkpoint_defer) {
+ flush_pending_checkpoints(false);
+ }
BEESLOGINFO("Saved beespoint.ini in " << save_time << "s");
}
+void
+BeesRoots::load_crawl_state()
+{
+ if (!m_roots_persistent) {
+ return;
+ }
+ if (!load_point()) {
+ import_legacy();
+ }
+}
+
+void
+BeesRoots::print_crawl_state(ostream &os)
+{
+ unique_lock<mutex> lock(m_mutex);
+ os << m_point_ini.write();
+}
+
void
BeesRoots::crawl_state_set_dirty()
{
});
return;
}
- m_stop_condvar.wait_for(lock, chrono::duration<double>(BEES_WRITEBACK_INTERVAL));
+ m_stop_condvar.wait_for(lock, chrono::duration<double>(m_checkpoint_interval));
}
}
#include <array>
#include <atomic>
+#include <deque>
#include <functional>
#include <list>
#include <mutex>
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.
+ /// A checkpoint snapshot waiting to be written to disk.
+ struct PendingCheckpoint {
+ uint64_t writeback_count; ///< m_hash_table->writeback_extent_count() at snapshot time.
+ string ini_string; ///< Serialised beespoint.ini content.
+ };
+ deque<PendingCheckpoint> m_pending_checkpoints; ///< Queue of deferred checkpoints.
vector<shared_ptr<BeesScanMode>> m_scanners; ///< Active scan-mode strategy objects.
/// state. If beescrawl.dat is also absent, both maps are left empty (first run).
/// beescrawl.dat is never modified.
void import_legacy();
- /// Write current m_extent_crawl_map and m_root_crawl_map to beespoint.ini.
- void save_point();
void crawl_state_set_dirty();
void crawl_state_erase(const BeesCrawlState &bcs);
bool all_crawlers_finished();
/// Return true if @p bcs.m_max_transid is already at or beyond the
/// configured transid bound.
bool up_to_date(const BeesCrawlState &bcs);
+ void snapshot_point();
+ void flush_pending_checkpoints(bool force);
friend class BeesCrawl;
friend class BeesScanMode;
/// Block until background threads have stopped.
void stop_wait();
+ /// Load beespoint.ini (or migrate from beescrawl.dat on first run).
+ void load_crawl_state();
+ /// Write current m_extent_crawl_map and m_root_crawl_map to beespoint.ini.
+ void save_point();
+ /// Serialize the current crawl state and write it to @p os in INI format.
+ void print_crawl_state(ostream &os);
+
/// Return true if the subvolume @p root is mounted read-only.
bool is_root_ro(uint64_t root);
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);
+ /// Write any remaining deferred checkpoints (call after hash table stop_wait() at shutdown).
+ void flush_deferred_checkpoints();
/// Return the lowest transaction ID seen across all active crawls.
uint64_t transid_min();
projects / bees / commitdiff