From 044b4c450082490114d488859c2cbcde971b4791 Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Fri, 21 Feb 2025 00:35:11 +0100 Subject: [PATCH] libutil: split retrySQLite into sync and async also mark the sync version as NeverAsync. a blocking wait on sqlite locks in a coroutine may never finish if it's a different coroutine on the same executor that is holding the lock, not another process. this propagates to the sqlite core interface, but no further. we'll assume that caches do not block on a database for very long, and we can't reasonably propagate never-async-ness out of stores unless we touch everything we'd touch for the async transition, again, twice. store code already assumes that it can block for however long it'll feel like that moment. we keep thread pools around for this reason. Change-Id: I62f77e1ac333cbe2e4e646dbcb1571463f2cf3fc --- lix/libexpr/eval-cache.cc | 2 +- lix/libfetchers/cache.cc | 3 +- lix/libstore/local-store.cc | 42 +++++++-------- lix/libstore/nar-info-disk-cache.cc | 19 +++---- lix/libstore/sqlite.cc | 9 ++-- lix/libstore/sqlite.hh | 61 ++++++++++++---------- tests/unit/libstore/nar-info-disk-cache.cc | 3 +- 7 files changed, 74 insertions(+), 65 deletions(-) diff --git a/lix/libexpr/eval-cache.cc b/lix/libexpr/eval-cache.cc index d75243d01..10040c1cc 100644 --- a/lix/libexpr/eval-cache.cc +++ b/lix/libexpr/eval-cache.cc @@ -45,7 +45,7 @@ struct AttrDb state->db = SQLite(dbPath); state->db.isCache(); - state->db.exec(schema); + state->db.exec(schema, always_progresses); state->insertAttribute = state->db.create( "insert or replace into Attributes(parent, name, type, value) values (?, ?, ?, ?)"); diff --git a/lix/libfetchers/cache.cc b/lix/libfetchers/cache.cc index 8b8773be9..0d9cc9e50 100644 --- a/lix/libfetchers/cache.cc +++ b/lix/libfetchers/cache.cc @@ -3,6 +3,7 @@ #include "lix/libutil/async.hh" #include "lix/libutil/sync.hh" #include "lix/libstore/store-api.hh" +#include "lix/libutil/types.hh" #include "lix/libutil/users.hh" #include @@ -50,7 +51,7 @@ struct CacheImpl : Cache state->db = SQLite(dbPath); state->db.isCache(); - state->db.exec(schema); + state->db.exec(schema, always_progresses); state->add = state->db.create( "insert or replace into Cache(input, info, path, immutable, timestamp) values (?, ?, ?, ?, ?)"); diff --git a/lix/libstore/local-store.cc b/lix/libstore/local-store.cc index f2d6dac7c..ce2276160 100644 --- a/lix/libstore/local-store.cc +++ b/lix/libstore/local-store.cc @@ -342,20 +342,20 @@ LocalStore::LocalStore(LocalStoreConfig config) if (curSchema < 8) { SQLiteTxn txn = state->db.beginTransaction(); - state->db.exec("alter table ValidPaths add column ultimate integer"); - state->db.exec("alter table ValidPaths add column sigs text"); + state->db.exec("alter table ValidPaths add column ultimate integer", always_progresses); + state->db.exec("alter table ValidPaths add column sigs text", always_progresses); txn.commit(); } if (curSchema < 9) { SQLiteTxn txn = state->db.beginTransaction(); - state->db.exec("drop table FailedPaths"); + state->db.exec("drop table FailedPaths", always_progresses); txn.commit(); } if (curSchema < 10) { SQLiteTxn txn = state->db.beginTransaction(); - state->db.exec("alter table ValidPaths add column ca text"); + state->db.exec("alter table ValidPaths add column ca text", always_progresses); txn.commit(); } @@ -525,7 +525,7 @@ void LocalStore::openDB(DBState & state, bool create) all. This can cause database corruption if the system crashes. */ std::string syncMode = settings.fsyncMetadata ? "normal" : "off"; - db.exec("pragma synchronous = " + syncMode); + db.exec("pragma synchronous = " + syncMode, always_progresses); /* Set the SQLite journal mode. WAL mode is fastest, so it's the default. */ @@ -538,7 +538,7 @@ void LocalStore::openDB(DBState & state, bool create) prevMode = use.getStr(0); } if (prevMode != mode) - db.exec("pragma main.journal_mode = " + mode + ";"); + db.exec("pragma main.journal_mode = " + mode + ";", always_progresses); if (mode == "wal" ) { /* persist the WAL files when the DB connection is closed. @@ -547,13 +547,13 @@ void LocalStore::openDB(DBState & state, bool create) * journal_size_limit to 2^40 bytes results in the WAL files getting * truncated to 0 on exit and limits the on disk size of the WAL files * to 2^40 bytes following a checkpoint */ - db.exec("pragma main.journal_size_limit = 1099511627776;"); + db.exec("pragma main.journal_size_limit = 1099511627776;", always_progresses); db.setPersistWAL(true); /* Increase the auto-checkpoint interval to 40000 pages. This seems enough to ensure that instantiating the NixOS system derivation is done in a single fsync(). */ - db.exec("pragma wal_autocheckpoint = 40000;"); + db.exec("pragma wal_autocheckpoint = 40000;", always_progresses); } /* Initialise the database schema, if necessary. */ @@ -561,7 +561,7 @@ void LocalStore::openDB(DBState & state, bool create) static const char schema[] = #include "schema.sql.gen.hh" ; - db.exec(schema); + db.exec(schema, always_progresses); } } @@ -812,7 +812,7 @@ void LocalStore::registerDrvOutput(const Realisation & info) (outputId.outputName) .exec(); } - }); + }, always_progresses); } void LocalStore::cacheDrvOutputMapping( @@ -885,7 +885,7 @@ std::shared_ptr LocalStore::queryPathInfoUncached(const Sto return retrySQLite([&]() { auto state(_dbState.lockSync(always_progresses)); return queryPathInfoInternal(*state, path); - }); + }, always_progresses); } @@ -973,7 +973,7 @@ bool LocalStore::isValidPathUncached(const StorePath & path) return retrySQLite([&]() { auto state(_dbState.lockSync(always_progresses)); return isValidPath_(*state, path); - }); + }, always_progresses); } @@ -994,7 +994,7 @@ StorePathSet LocalStore::queryAllValidPaths() StorePathSet res; while (use.next()) res.insert(parseStorePath(use.getStr(0))); return res; - }); + }, always_progresses); } @@ -1012,7 +1012,7 @@ void LocalStore::queryReferrers(const StorePath & path, StorePathSet & referrers return retrySQLite([&]() { auto state(_dbState.lockSync(always_progresses)); queryReferrers(*state, path, referrers); - }); + }, always_progresses); } @@ -1028,7 +1028,7 @@ StorePathSet LocalStore::queryValidDerivers(const StorePath & path) derivers.insert(parseStorePath(useQueryValidDerivers.getStr(1))); return derivers; - }); + }, always_progresses); } @@ -1046,7 +1046,7 @@ LocalStore::queryStaticPartialDerivationOutputMap(const StorePath & path) use.getStr(0), parseStorePath(use.getStr(1))); return outputs; - }); + }, always_progresses); } std::optional LocalStore::queryPathFromHashPart(const std::string & hashPart) @@ -1066,7 +1066,7 @@ std::optional LocalStore::queryPathFromHashPart(const std::string & h if (s.has_value() && s->starts_with(prefix)) return parseStorePath(*s); return {}; - }); + }, always_progresses); } @@ -1164,7 +1164,7 @@ void LocalStore::registerValidPaths(const ValidPathInfos & infos) }}); txn.commit(); - }); + }, always_progresses); } @@ -1524,7 +1524,7 @@ void LocalStore::invalidatePathChecked(const StorePath & path) } txn.commit(); - }); + }, always_progresses); } @@ -1733,7 +1733,7 @@ void LocalStore::addSignatures(const StorePath & storePath, const StringSet & si updatePathInfo(*state, *info); txn.commit(); - }); + }, always_progresses); } @@ -1821,7 +1821,7 @@ std::shared_ptr LocalStore::queryRealisationUncached(const Dr auto maybeRealisation = retrySQLite([&]() { auto state(_dbState.lockSync(always_progresses)); return queryRealisation_(*state, id); - }); + }, always_progresses); if (maybeRealisation) return std::make_shared(maybeRealisation.value()); else diff --git a/lix/libstore/nar-info-disk-cache.cc b/lix/libstore/nar-info-disk-cache.cc index c31ae33c6..b3b13a121 100644 --- a/lix/libstore/nar-info-disk-cache.cc +++ b/lix/libstore/nar-info-disk-cache.cc @@ -3,6 +3,7 @@ #include "lix/libutil/sync.hh" #include "lix/libstore/sqlite.hh" #include "lix/libstore/globals.hh" +#include "lix/libutil/types.hh" #include "lix/libutil/users.hh" #include "lix/libutil/strings.hh" @@ -97,7 +98,7 @@ public: state->db.isCache(); - state->db.exec(schema); + state->db.exec(schema, always_progresses); state->insertCache = state->db.create( "insert into BinaryCaches(url, timestamp, storeDir, wantMassQuery, priority) values (?1, ?2, ?3, ?4, ?5) on conflict (url) do update set timestamp = ?2, storeDir = ?3, wantMassQuery = ?4, priority = ?5 returning id;"); @@ -158,7 +159,7 @@ public: "insert or replace into LastPurge(dummy, value) values ('', ?)") .use()(now).exec(); } - }); + }, always_progresses); } Cache & getCache(State & state, const std::string & uri) @@ -219,7 +220,7 @@ public: txn.commit(); return ret.id; - }); + }, always_progresses); } std::optional upToDateCacheExists(const std::string & uri) override @@ -234,7 +235,7 @@ public: .wantMassQuery = cache->wantMassQuery, .priority = cache->priority }; - }); + }, always_progresses); } std::pair> lookupNarInfo( @@ -278,7 +279,7 @@ public: narInfo->ca = ContentAddress::parseOpt(queryNAR.getStr(11)); return {oValid, narInfo}; - }); + }, always_progresses); } std::pair> lookupRealisation( @@ -309,7 +310,7 @@ public: "Local disk cache")); return {oValid, realisation}; - }); + }, always_progresses); } void upsertNarInfo( @@ -349,7 +350,7 @@ public: (hashPart) (time(0)).exec(); } - }); + }, always_progresses); } void upsertRealisation( @@ -366,7 +367,7 @@ public: (realisation.id.to_string()) (realisation.toJSON().dump()) (time(0)).exec(); - }); + }, always_progresses); } @@ -382,7 +383,7 @@ public: (cache.id) (id.to_string()) (time(0)).exec(); - }); + }, always_progresses); } }; diff --git a/lix/libstore/sqlite.cc b/lix/libstore/sqlite.cc index a045a102f..4ef929b06 100644 --- a/lix/libstore/sqlite.cc +++ b/lix/libstore/sqlite.cc @@ -5,6 +5,7 @@ #include "lix/libutil/logging.hh" #include "lix/libutil/result.hh" #include "lix/libutil/signals.hh" +#include "lix/libutil/types.hh" #include "lix/libutil/url.hh" #include @@ -85,7 +86,7 @@ SQLite::SQLite(const Path & path, SQLiteOpenMode mode) sqlite3_trace(db, &traceSQL, nullptr); } - exec("pragma foreign_keys = 1"); + exec("pragma foreign_keys = 1", always_progresses); } void SQLite::Close::operator()(sqlite3 * db) @@ -100,11 +101,11 @@ void SQLite::Close::operator()(sqlite3 * db) void SQLite::isCache() { - exec("pragma synchronous = off"); - exec("pragma main.journal_mode = truncate"); + exec("pragma synchronous = off", always_progresses); + exec("pragma main.journal_mode = truncate", always_progresses); } -void SQLite::exec(const std::string & stmt) +void SQLite::exec(const std::string & stmt, NeverAsync) { retrySQLite([&]() { if (sqlite3_exec(db.get(), stmt.c_str(), 0, 0, 0) != SQLITE_OK) diff --git a/lix/libstore/sqlite.hh b/lix/libstore/sqlite.hh index 9098ab078..5d3aa52ef 100644 --- a/lix/libstore/sqlite.hh +++ b/lix/libstore/sqlite.hh @@ -86,7 +86,7 @@ public: */ void isCache(); - void exec(const std::string & stmt); + void exec(const std::string & stmt, NeverAsync = {}); SQLiteStmt create(const std::string & stmt); @@ -223,38 +223,43 @@ kj::Promise> handleSQLiteBusyAsync(const SQLiteBusy & e, std::chron * database is busy. */ template -auto retrySQLite(F fun) + requires(!requires(F f) { [](kj::Promise) {}(f()); }) +auto retrySQLite(F fun, NeverAsync = {}) { auto nextWarning = std::chrono::steady_clock::now() + std::chrono::seconds(1); - if constexpr (requires (F f) { [](kj::Promise>){}(f()); }) { - return [](std::chrono::time_point nextWarning, F fun) -> decltype(fun()) { - while (true) { - kj::Promise> handleBusy{nullptr}; - try { - if constexpr (std::is_same_v>>) { - LIX_TRY_AWAIT(fun()); - co_return result::success(); - } else { - co_return LIX_TRY_AWAIT(fun()); - } - } catch (SQLiteBusy & e) { - handleBusy = handleSQLiteBusyAsync(e, nextWarning); - } catch (...) { - co_return result::current_exception(); - } - LIX_TRY_AWAIT(handleBusy); - } - }(nextWarning, std::move(fun)); - } else { - while (true) { - try { - return fun(); - } catch (SQLiteBusy & e) { - handleSQLiteBusy(e, nextWarning); - } + while (true) { + try { + return fun(); + } catch (SQLiteBusy & e) { + handleSQLiteBusy(e, nextWarning); } } } +template + requires requires(F f) { [](kj::Promise>) {}(f()); } +auto retrySQLite(F fun) +{ + return [](F fun) -> decltype(fun()) { + auto nextWarning = std::chrono::steady_clock::now() + std::chrono::seconds(1); + + while (true) { + kj::Promise> handleBusy{nullptr}; + try { + if constexpr (std::is_same_v>>) { + LIX_TRY_AWAIT(fun()); + co_return result::success(); + } else { + co_return LIX_TRY_AWAIT(fun()); + } + } catch (SQLiteBusy & e) { + handleBusy = handleSQLiteBusyAsync(e, nextWarning); + } catch (...) { + co_return result::current_exception(); + } + LIX_TRY_AWAIT(handleBusy); + } + }(std::move(fun)); +} } diff --git a/tests/unit/libstore/nar-info-disk-cache.cc b/tests/unit/libstore/nar-info-disk-cache.cc index 741c9cb53..5e49a34bd 100644 --- a/tests/unit/libstore/nar-info-disk-cache.cc +++ b/tests/unit/libstore/nar-info-disk-cache.cc @@ -4,6 +4,7 @@ #include #include "lix/libstore/sqlite.hh" #include "lix/libstore/temporary-dir.hh" +#include "lix/libutil/types.hh" #include @@ -59,7 +60,7 @@ TEST(NarInfoDiskCacheImpl, create_and_read) { } // Pretend that the caches are older, but keep one up to date, as "background noise" - db.exec("update BinaryCaches set timestamp = timestamp - 1 - 7 * 24 * 3600 where url <> 'https://xyz';"); + db.exec("update BinaryCaches set timestamp = timestamp - 1 - 7 * 24 * 3600 where url <> 'https://xyz';", always_progresses); // This shows that the in-memory cache works {