From 7d60d2728afd8da0bb54f49f74ea0c8dfd78c673 Mon Sep 17 00:00:00 2001 From: "Dirk-Jan C. Binnema" Date: Tue, 21 Jul 2026 09:18:54 +0300 Subject: [PATCH] index: generalize pre-fetch paths We already pre-fetched db file paths for the cleanup-case (since cdb619e4f), now let's generalize this to all indexing. For now this is mostly performance-neutral (lazy-rescan is slightly faster); however, this simplifies the code. --- lib/mu-indexer.cc | 304 ++++++++++++++++--------------------- lib/mu-store.cc | 22 --- lib/mu-store.hh | 20 --- lib/mu-xapian-db.hh | 16 +- lib/tests/test-mu-store.cc | 3 + 5 files changed, 145 insertions(+), 220 deletions(-) diff --git a/lib/mu-indexer.cc b/lib/mu-indexer.cc index daf84d72..7ea94d9a 100644 --- a/lib/mu-indexer.cc +++ b/lib/mu-indexer.cc @@ -26,8 +26,6 @@ #include #include #include -#include -#include #include #include #include @@ -109,12 +107,15 @@ struct Indexer::Private { void maybe_start_worker(); + void prefetch_path_terms(bool for_cleanup); + bool mark_seen(const std::string& path); + void mark_seen_dir(const std::string& dir_path); + void scan_worker(); bool add_message(const std::string& path); - void cleanup_from_scratch(); - void cleanup_incremental(); + void cleanup(); bool start(const Indexer::Config& conf, bool block); bool stop(); @@ -148,12 +149,71 @@ struct Indexer::Private { uint64_t last_index_{}; - // pathnames we've seen traversing the maildir hierarchy. entries - // with a trailing slash are maildir directories we've skipped as - // up-to-date. - std::vector seen_maildir_paths_; + using PathTerm = std::pair; + std::vector db_path_terms_; + /**< all path-terms in the store (at scan-start), + in Xapian's (ascending) and a seen marker */ + bool use_db_path_terms_{}; }; +void +Indexer::Private::prefetch_path_terms(bool for_cleanup) +{ + use_db_path_terms_ = false; + db_path_terms_.clear(); + + // The in-memory path-terms serve two purposes: + // + // i. checking whether some message is already in the store + // without db access + // ii. unless we need it for cleanup: whatever + // is left unmarked after scanning is an orphan. + if (!for_cleanup && conf_.lazy_check) + return; + + db_path_terms_.reserve(store_.size()); + store_.for_each_term(Field::Id::Path, [&](const std::string& term) { + db_path_terms_.emplace_back(term, false/*!seen*/); + return true; + }); + use_db_path_terms_ = true; + + mu_debug("prefetched {} path-term(s)", db_path_terms_.size()); +} + +bool +Indexer::Private::mark_seen(const std::string& path) +{ + if (!use_db_path_terms_) + return false; + + // N.B. Xapian yields terms in ascending byte-lexicographic order, so + // db_path_terms_ is sorted and we can binary-search. + const auto term{field_from_id(Field::Id::Path).xapian_term(path)}; + const auto it{std::ranges::lower_bound(db_path_terms_, term, {}, + &PathTerm::first)}; + if (it == db_path_terms_.end() || it->first != term) + return false; // not in the store. + + it->second = true; + return true; +} + +void +Indexer::Private::mark_seen_dir(const std::string& dir_path) +{ + if (!use_db_path_terms_) + return; + + // mark all store messages under dir_path as seen + const auto prefix{field_from_id(Field::Id::Path) + .xapian_term(dir_path + "/")}; + for (auto it = std::ranges::lower_bound(db_path_terms_, prefix, {}, + &PathTerm::first); + it != db_path_terms_.end() && it->first.starts_with(prefix); ++it) + it->second = true; +} + bool Indexer::Private::handler(const std::string& fullpath, struct stat* statbuf, Scanner::HandleType htype) @@ -179,7 +239,7 @@ Indexer::Private::handler(const std::string& fullpath, struct stat* statbuf, htype == Scanner::HandleType::EnterNewCur) { mu_debug("skip {} (seems up-to-date: {:%FT%T} >= {:%FT%T})", fullpath, mu_time(dirstamp_), mu_time(statbuf->st_ctime)); - seen_maildir_paths_.emplace_back(fullpath + "/"); + mark_seen_dir(fullpath); return false; } @@ -196,7 +256,7 @@ Indexer::Private::handler(const std::string& fullpath, struct stat* statbuf, auto noupdate = ::access((fullpath + "/.noupdate").c_str(), F_OK) == 0; if (noupdate) { mu_debug("skip {} (has .noupdate)", fullpath); - seen_maildir_paths_.emplace_back(fullpath + "/"); + mark_seen_dir(fullpath); return false; } } @@ -205,13 +265,21 @@ Indexer::Private::handler(const std::string& fullpath, struct stat* statbuf, return true; } case Scanner::HandleType::LeaveDir: { - handle_item({fullpath, WorkItem::Type::Dir}); + // don't touch dirstamps in a cleanup-only run: nothing was + // indexed, so the dir is not to be considered up-to-date. + if (conf_.scan) + handle_item({fullpath, WorkItem::Type::Dir}); return true; } case Scanner::HandleType::File: { ++progress_.checked; - seen_maildir_paths_.push_back(fullpath); + // this file is present in the file-system; also remember + // whether it is in the store already. + const auto in_store{mark_seen(fullpath)}; + + if (!conf_.scan) + return false; // cleanup-only run; only mark. if (conf_.lazy_check && static_cast(statbuf->st_ctime) < last_index_) { // in lazy mode, ignore the file if it has not changed @@ -225,7 +293,8 @@ Indexer::Private::handler(const std::string& fullpath, struct stat* statbuf, } // if the message is not in the db yet, or not up-to-date, queue // it for updating/inserting. - if (statbuf->st_ctime <= dirstamp_ && store_.contains_message(fullpath)) + if (statbuf->st_ctime <= dirstamp_ && + (use_db_path_terms_ ? in_store : store_.contains_message(fullpath))) return false; handle_item({fullpath, WorkItem::Type::File}); @@ -288,128 +357,21 @@ Indexer::Private::handle_item(WorkItem&& item) } void -Indexer::Private::cleanup_from_scratch() -{ - mu_debug("starting cleanup without using scan results"); - - std::vector orphans; // store messages without files. - - using DirFiles = std::unordered_set; - std::unordered_map dir_cache; - - // get a set of file names in this directory. - const auto get_dir_files = [](const std::string& path) -> DirFiles { - DirFiles ret; - if (auto dir{::opendir(path.c_str())}; dir) { - dirent* dentry{}; - while ((dentry = ::readdir(dir))) { - ret.emplace(dentry->d_name); - } - ::closedir(dir); - } - - return ret; - }; - - // is the file present in our set? - const auto is_file_present = [&](const std::string& path) -> bool { - std::string dir = dirname(path); - auto [it, inserted] = dir_cache.try_emplace(dir); - DirFiles& dir_files = it->second; - if (inserted) { - dir_files = get_dir_files(dir); - } - return dir_files.find(basename(path)) != dir_files.end(); - }; - - store_.for_each_message_path([&](Store::Id id, const std::string& path) { - if (!is_file_present(path)) { - mu_debug("cannot read {} (id={}); queuing for removal from store", - path, id); - orphans.emplace_back(id); - } - - return state_ == IndexState::Cleaning; - }); - - if (orphans.empty()) - mu_debug("nothing to clean up"); - else { - mu_debug("removing {} stale message(s) from store", orphans.size()); - store_.remove_messages(orphans); - progress_.removed += orphans.size(); - } -} - -void -Indexer::Private::cleanup_incremental() +Indexer::Private::cleanup() { mu_debug("starting cleanup after scan"); - // Sort the seen paths into the same order Xapian will give its terms to us - std::vector fs_terms; - fs_terms.reserve(seen_maildir_paths_.size()); - for (std::string_view fullpath : seen_maildir_paths_) - fs_terms.emplace_back(field_from_id(Field::Id::Path).xapian_term(fullpath)); - std::ranges::sort(fs_terms); - - // Discard duplicates from fs_terms in case two paths collided to one term, e.g. over - // case. That shouldn't happen, but be correct-ish if it does. - auto [new_end, old_end] = std::ranges::unique(fs_terms); - if (new_end != old_end) { - mu_warning("collisions under term normalization: using regular cleanup"); - cleanup_from_scratch(); - return; - } - - fs_terms.erase(new_end, old_end); - seen_maildir_paths_.clear(); - - // Walk through all the path terms. If the DB has a path term that we didn't see in our - // filesystem walk above, add it the orphans list for removal from the DB. If we were in - // lazy scan mode, we may have skipped some directories entirely: these are represented by - // entries in fs_terms with a trailing slash. When we see one, we deem all DB entries - // that have the skip entry as a prefix as present. - - size_t fs_terms_pos = 0; - auto current_fs_term = [&]() -> std::string_view { - if (fs_terms_pos < fs_terms.size()) - return fs_terms[fs_terms_pos]; - // N.B. '~' compares greater than the start of any field shortcut, so use it as an - // after-the-end sentinel. - return "~"; - }; - + // during the scan, each file (and each wholesale-skipped + // directory) marked its path-term; whatever is left unmarked is + // an orphan: a message in the store without a file in the maildir. + // + // N.B. this snapshot was taken at scan-start, so messages that + // appeared in the store _during_ the scan are not in it, and thus + // cannot be orphaned by mistake. std::vector orphan_terms; - - auto handle_db_term = [&](std::string_view db_term) { - for (;;) { - std::string_view fs_term = current_fs_term(); - bool is_wildcard = fs_term.ends_with('/'); - if (is_wildcard && db_term.starts_with(fs_term)) { - return true; - } - int cmp = db_term.compare(fs_term); - if (cmp < 0) { - mu_debug("orphan in db={} but not fs={}", db_term, fs_term); - orphan_terms.emplace_back(db_term); - return true; - } - - if (cmp == 0) { - ++fs_terms_pos; - return true; - } - - ++fs_terms_pos; - // FS has an entry not in the DB. If not a directory, we should have - // indexed it. - if (!is_wildcard) - mu_warning("unexpectedly unindexed message: {}", fs_term); - } - }; - - store_.for_each_term(Field::Id::Path, handle_db_term); + for (auto&& [term, seen] : db_path_terms_) + if (!seen) + orphan_terms.emplace_back(std::move(term)); if (orphan_terms.empty()) mu_debug("nothing to clean up"); @@ -430,10 +392,13 @@ void Indexer::Private::scan_worker() { progress_.reset(); - seen_maildir_paths_.clear(); started_ = time(NULL); - if (conf_.scan) { + // cleanup gets its orphans by marking the in-memory path-terms + // during the file-system scan, so we need the scanner even for a + // cleanup-only run. + if (conf_.scan || conf_.cleanup) { + prefetch_path_terms(conf_.cleanup); mu_debug("starting scanner"); if (!scanner_.start()) { // blocks. mu_warning("failed to start scanner"); @@ -443,51 +408,23 @@ Indexer::Private::scan_worker() mu_debug("scanner finished"); } - enum class CleanupKind { - None, - FromScratch, - Incremental, - }; - - CleanupKind cleanup_kind = CleanupKind::Incremental; bool aborted = state_ == IndexState::Aborting; - if (cleanup_kind >= CleanupKind::None && !conf_.cleanup) { - mu_debug("cleanup: not running as requested"); - cleanup_kind = CleanupKind::None; - } - - if (cleanup_kind > CleanupKind::None && aborted) { - mu_debug("cleanup: disabling because indexer aborted"); - cleanup_kind = CleanupKind::None; - } - - if (cleanup_kind >= CleanupKind::Incremental && - g_getenv("MU_NO_INCREMENTAL_CLEANUP")) { - mu_debug("cleanup: not using incremental: MU_NO_INCREMENTAL_CLEANUP in environ"); - cleanup_kind = CleanupKind::FromScratch; - } - - if (cleanup_kind >= CleanupKind::Incremental && !conf_.scan) { - mu_debug("cleanup: not using incremental: scan not done"); - cleanup_kind = CleanupKind::FromScratch; - } - state_.change_to(IndexState::Cleaning); - switch (cleanup_kind) { - case CleanupKind::None: - break; - case CleanupKind::FromScratch: - cleanup_from_scratch(); - break; - case CleanupKind::Incremental: - cleanup_incremental(); - break; - } + if (!conf_.cleanup) + mu_debug("cleanup: not running as requested"); + else if (aborted) + mu_debug("cleanup: disabling because indexer aborted"); + else + cleanup(); + + // release the in-memory path-terms. + use_db_path_terms_ = false; + db_path_terms_ = {}; aborted = state_ == IndexState::Aborting; - if (!aborted) { + if (!aborted && conf_.scan) { // Store started time, not ending time, so that next time we run we know to scan // anything that appeared during our scan. store_.config().set(started_.value()); @@ -507,7 +444,11 @@ Indexer::Private::start(const Indexer::Config& conf, bool block) conf_ = conf; - if (store_.empty() && conf_.lazy_check) { + // refresh: the indexer may be re-used for multiple runs, and the + // add_document fast-path is only valid while the store is empty. + was_empty_ = store_.empty(); + + if (was_empty_ && conf_.lazy_check) { mu_debug("turn off lazy check since we have an empty store"); conf_.lazy_check = false; } @@ -768,6 +709,23 @@ test_index_cleanup() g_assert_false(idx.is_running()); g_assert_true(idx.stop()); g_assert_cmpuint(store->size(),==, 13); + + // remove another message + { + auto mpath = join_paths(mdir, "bar", "cur", "mail5"); + auto res = run_command({"rm", mpath}); + assert_valid_result(res); + g_assert_cmpuint(res->exit_code,==, 0); + } + + // cleanup-only run (no scan); message is gone from store. + conf.scan = false; + g_assert_true(idx.start(conf)); + while (idx.is_running()) + g_usleep(10000); + g_assert_false(idx.is_running()); + g_assert_true(idx.stop()); + g_assert_cmpuint(store->size(),==, 12); } diff --git a/lib/mu-store.cc b/lib/mu-store.cc index 81a34334..64604748 100644 --- a/lib/mu-store.cc +++ b/lib/mu-store.cc @@ -756,28 +756,6 @@ Store::label_map() const return priv_->labels_cache_.label_map(); } -std::size_t -Store::for_each_message_path(Store::ForEachMessageFunc msg_func) const -{ - size_t n{}; - - xapian_try([&] { - std::lock_guard guard{priv_->lock_}; - auto enq{xapian_db().enquire()}; - - enq.set_query(Xapian::Query::MatchAll); - enq.set_cutoff(0, 0); - - Xapian::MSet matches(enq.get_mset(0, xapian_db().size())); - constexpr auto path_no{field_from_id(Field::Id::Path).value_no()}; - for (auto&& it = matches.begin(); it != matches.end(); ++it, ++n) - if (!msg_func(*it, it.get_document().get_value(path_no))) - break; - }); - - return n; -} - std::size_t Store::for_each_term(Field::Id field_id, Store::ForEachTermFunc func) const { diff --git a/lib/mu-store.hh b/lib/mu-store.hh index 03f26ce0..27022413 100644 --- a/lib/mu-store.hh +++ b/lib/mu-store.hh @@ -413,26 +413,6 @@ public: */ LabelsCache::Map label_map() const; - /** - * Prototype for the ForEachMessageFunc - * - * @param id :t store Id for the message - * @param path: the absolute path to the message - * - * @return true if for_each should continue; false to quit - */ - using ForEachMessageFunc = std::function; - - /** - * Call @param func for each document in the store. This takes a lock on - * the store, so the func should _not_ call any other Store:: methods. - * - * @param func a Callable invoked for each message. - * - * @return the number of times func was invoked - */ - size_t for_each_message_path(ForEachMessageFunc func) const; - /** * Prototype for the ForEachTermFunc * diff --git a/lib/mu-xapian-db.hh b/lib/mu-xapian-db.hh index c7e1f631..3404c5f4 100644 --- a/lib/mu-xapian-db.hh +++ b/lib/mu-xapian-db.hh @@ -359,7 +359,7 @@ public: Result add_document(const Xapian::Document& doc) { return xapian_try_result([&]{ auto&& id{wdb().add_document(doc)}; - set_timestamp(MetadataIface::last_change_key); + dirty_ = true; maybe_commit(); return Ok(std::move(id)); }); @@ -379,7 +379,7 @@ public: const Xapian::Document& doc) { return xapian_try_result([&]{ auto&& id{wdb().replace_document(term, doc)}; - set_timestamp(MetadataIface::last_change_key); + dirty_ = true; maybe_commit(); return Ok(std::move(id)); }); @@ -389,7 +389,7 @@ public: const Xapian::Document& doc) { return xapian_try_result([&]{ wdb().replace_document(id, doc); - set_timestamp(MetadataIface::last_change_key); + dirty_ = true; maybe_commit(); return Ok(std::move(id)); }); @@ -405,7 +405,7 @@ public: Result delete_document(const std::string& term) { return xapian_try_result([&]{ wdb().delete_document(term); - set_timestamp(MetadataIface::last_change_key); + dirty_ = true; maybe_commit(); return Ok(); }); @@ -413,7 +413,7 @@ public: Result delete_document(Xapian::docid id) { return xapian_try_result([&]{ wdb().delete_document(id); - set_timestamp(MetadataIface::last_change_key); + dirty_ = true; maybe_commit(); return Ok(); }); @@ -486,6 +486,11 @@ private: "forced={}", changes_, in_transaction() ? "yes" : "no", force ? "yes" : "no"); + // record the last doc change as part of commit + if (dirty_) { + set_timestamp(MetadataIface::last_change_key); + dirty_ = false; + } if (in_transaction()) { db.commit_transaction(); in_transaction_ = {}; @@ -515,6 +520,7 @@ private: std::string path_; DbType db_; size_t changes_{}; + bool dirty_{}; /* document changes not yet timestamped? */ bool in_transaction_{}; size_t batch_size_; }; diff --git a/lib/tests/test-mu-store.cc b/lib/tests/test-mu-store.cc index 5307aa5f..b4db62e9 100644 --- a/lib/tests/test-mu-store.cc +++ b/lib/tests/test-mu-store.cc @@ -347,6 +347,9 @@ World! // term != msg2->document().xapian_document().termlist_end(); ++term) // g_message(">>> %s", (*term).c_str()); + // last-change is recorded as part of a commit. + store->xapian_db().request_commit(true/*force*/); + const auto stats{store->statistics()}; g_assert_cmpuint(stats.size,==,store->size()); g_assert_cmpuint(stats.last_index,==,0);