libstore: asyncify openStore

Change-Id: Ia152bfd2014851590328e732f434dff6695304bd

lix/legacy/build-remote.cc

@@ -84,7 +84,7 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings
 
         initPlugins();
 
-        auto store = openStore();
+        auto store = aio.blockOn(openStore());
 
         /* It would be more appropriate to use $XDG_RUNTIME_DIR, since
            that gets cleared on reboot, but it wouldn't work on macOS. */
@@ -243,7 +243,7 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings
 
                     Activity act(*logger, lvlTalkative, actUnknown, fmt("connecting to '%s'", bestMachine->storeUri));
 
-                    sshStore = bestMachine->openStore();
+                    sshStore = aio.blockOn(bestMachine->openStore());
                     sshStore->connect();
                     storeUri = bestMachine->storeUri;
 

lix/legacy/nix-build.cc

@@ -195,8 +195,8 @@ static void main_nix_build(AsyncIoRoot & aio, std::string programName, Strings a
     if (outLink.empty())
         outLink = (Path) tmpDir + "/result";
 
-    auto store = openStore();
-    auto evalStore = myArgs.evalStoreUrl ? openStore(*myArgs.evalStoreUrl) : store;
+    auto store = aio.blockOn(openStore());
+    auto evalStore = myArgs.evalStoreUrl ? aio.blockOn(openStore(*myArgs.evalStoreUrl)) : store;
 
     auto evaluator = std::make_unique<Evaluator>(aio, myArgs.searchPath, evalStore, store);
     evaluator->repair = myArgs.repair;

lix/legacy/nix-channel.cc

@@ -7,6 +7,7 @@
 #include "lix/libfetchers/fetchers.hh"
 #include "lix/libexpr/eval-settings.hh" // for defexpr
 #include "lix/libstore/temporary-dir.hh"
+#include "lix/libutil/async.hh"
 #include "lix/libutil/users.hh"
 #include "nix-channel.hh"
 
@@ -75,11 +76,11 @@ static void removeChannel(const std::string & name)
 static Path nixDefExpr;
 
 // Fetch Nix expressions and binary cache URLs from the subscribed channels.
-static void update(const StringSet & channelNames)
+static void update(AsyncIoRoot & aio, const StringSet & channelNames)
 {
     readChannels();
 
-    auto store = openStore();
+    auto store = aio.blockOn(openStore());
 
     auto [fd, unpackChannelPath] = createTempFile();
     writeFull(fd.get(),
@@ -241,7 +242,7 @@ static int main_nix_channel(AsyncIoRoot & aio, std::string programName, Strings
                     std::cout << channel.first << ' ' << channel.second << '\n';
                 break;
             case cUpdate:
-                update(StringSet(args.begin(), args.end()));
+                update(aio, StringSet(args.begin(), args.end()));
                 break;
             case cListGenerations:
                 if (!args.empty())

lix/legacy/nix-collect-garbage.cc

@@ -94,7 +94,7 @@ static int main_nix_collect_garbage(AsyncIoRoot & aio, std::string programName,
         } else {
             options.action = GCOptions::gcReturnDead;
         }
-        auto store = openStore();
+        auto store = aio.blockOn(openStore());
         auto & gcStore = require<GcStore>(*store);
         GCResults results;
         PrintFreed freed(true, results);

lix/legacy/nix-copy-closure.cc

@@ -48,8 +48,8 @@ static int main_nix_copy_closure(AsyncIoRoot & aio, std::string programName, Str
             throw UsageError("no host name specified");
 
         auto remoteUri = "ssh://" + sshHost + (gzip ? "?compress=true" : "");
-        auto to = toMode ? openStore(remoteUri) : openStore();
-        auto from = toMode ? openStore() : openStore(remoteUri);
+        auto to = aio.blockOn(toMode ? openStore(remoteUri) : openStore());
+        auto from = aio.blockOn(toMode ? openStore() : openStore(remoteUri));
 
         RealisedPath::Set storePaths2;
         for (auto & path : storePaths)

lix/legacy/nix-env.cc

@@ -1539,7 +1539,7 @@ static int main_nix_env(AsyncIoRoot & aio, std::string programName, Strings argv
         if (showHelp) showManPage("nix-env" + opName);
         if (!op) throw UsageError("no operation specified");
 
-        auto store = openStore();
+        auto store = aio.blockOn(openStore());
 
         globals.state = std::make_shared<Evaluator>(aio, myArgs.searchPath, store);
         globals.state->repair = myArgs.repair;

lix/legacy/nix-instantiate.cc

@@ -154,8 +154,8 @@ static int main_nix_instantiate(AsyncIoRoot & aio, std::string programName, Stri
         if (evalOnly && !wantsReadWrite)
             settings.readOnlyMode = true;
 
-        auto store = openStore();
-        auto evalStore = myArgs.evalStoreUrl ? openStore(*myArgs.evalStoreUrl) : store;
+        auto store = aio.blockOn(openStore());
+        auto evalStore = myArgs.evalStoreUrl ? aio.blockOn(openStore(*myArgs.evalStoreUrl)) : store;
 
         auto evaluator = std::make_unique<Evaluator>(aio, myArgs.searchPath, evalStore, store);
         auto state = evaluator->begin(aio);

lix/legacy/nix-store.cc

@@ -1173,7 +1173,7 @@ static int main_nix_store(AsyncIoRoot & aio, std::string programName, Strings ar
         if (!op) throw UsageError("no operation specified");
 
         if (op != opDump && op != opRestore) /* !!! hack */
-            store = openStore();
+            store = aio.blockOn(openStore());
 
         op(aio, std::move(opFlags), std::move(opArgs));
 

lix/libcmd/command.cc

@@ -41,7 +41,7 @@ ref<Store> StoreCommand::getStore()
 
 ref<Store> StoreCommand::createStore()
 {
-    return openStore();
+    return aio().blockOn(openStore());
 }
 
 void StoreCommand::run()
@@ -68,7 +68,7 @@ CopyCommand::CopyCommand()
 
 ref<Store> CopyCommand::createStore()
 {
-    return srcUri.empty() ? StoreCommand::createStore() : openStore(srcUri);
+    return srcUri.empty() ? StoreCommand::createStore() : aio().blockOn(openStore(srcUri));
 }
 
 ref<Store> CopyCommand::getDstStore()
@@ -76,7 +76,7 @@ ref<Store> CopyCommand::getDstStore()
     if (srcUri.empty() && dstUri.empty())
         throw UsageError("you must pass '--from' and/or '--to'");
 
-    return dstUri.empty() ? openStore() : openStore(dstUri);
+    return aio().blockOn(dstUri.empty() ? openStore() : openStore(dstUri));
 }
 
 EvalCommand::EvalCommand()
@@ -98,7 +98,7 @@ EvalCommand::~EvalCommand()
 ref<Store> EvalCommand::getEvalStore()
 {
     if (!evalStore)
-        evalStore = evalStoreUrl ? openStore(*evalStoreUrl) : getStore();
+        evalStore = evalStoreUrl ? aio().blockOn(openStore(*evalStoreUrl)) : getStore();
     return ref<Store>(evalStore);
 }
 

lix/libcmd/common-eval-args.cc

@@ -155,7 +155,7 @@ MixEvalArgs::MixEvalArgs()
             fetchers::overrideRegistry(from.input, to.input, extraAttrs);
         }},
         .completer = {[&](AddCompletions & completions, size_t, std::string_view prefix) {
-            completeFlakeRef(completions, openStore(), prefix);
+            completeFlakeRef(completions, aio().blockOn(openStore()), prefix);
         }}
     });
 

lix/libcmd/repl.cc

@@ -1152,7 +1152,12 @@ ReplExitStatus AbstractNixRepl::run(
 ReplExitStatus AbstractNixRepl::runSimple(EvalState & evalState, const ValMap & extraEnv)
 {
     return run(
-        {}, openStore(), evalState, [] {  return AnnotatedValues{}; }, extraEnv, nullptr
+        {},
+        evalState.aio.blockOn(openStore()),
+        evalState,
+        [] { return AnnotatedValues{}; },
+        extraEnv,
+        nullptr
     );
 }
 

lix/libexpr/primops/fetchClosure.cc

@@ -198,7 +198,7 @@ void prim_fetchClosure(EvalState & state, const PosIdx pos, Value * * args, Valu
             .pos = state.ctx.positions[pos]
         });
 
-    auto fromStore = openStore(parsedURL.to_string());
+    auto fromStore = state.aio.blockOn(openStore(parsedURL.to_string()));
 
     if (toPath)
         runFetchClosureWithRewrite(state, pos, *fromStore, *fromPath, *toPath, v);

lix/libstore/machines.cc

@@ -1,6 +1,7 @@
 #include "lix/libstore/machines.hh"
 #include "lix/libstore/globals.hh"
 #include "lix/libstore/store-api.hh"
+#include "lix/libutil/async.hh"
 #include "lix/libutil/strings.hh"
 
 #include <algorithm>
@@ -64,8 +65,8 @@ bool Machine::mandatoryMet(const std::set<std::string> & features) const
         });
 }
 
-ref<Store> Machine::openStore() const
-{
+kj::Promise<Result<ref<Store>>> Machine::openStore() const
+try {
     StoreConfig::Params storeParams;
     if (storeUri.starts_with("ssh://")) {
         storeParams["log-fd"] = "4";
@@ -91,7 +92,9 @@ ref<Store> Machine::openStore() const
         append(mandatoryFeatures);
     }
 
-    return nix::openStore(storeUri, storeParams);
+    co_return TRY_AWAIT(nix::openStore(storeUri, storeParams));
+} catch (...) {
+    co_return result::current_exception();
 }
 
 static std::vector<std::string> expandBuilderLines(const std::string & builders)

lix/libstore/machines.hh

@@ -2,6 +2,8 @@
 ///@file
 
 #include "lix/libutil/ref.hh"
+#include "lix/libutil/result.hh"
+#include <kj/async.h>
 #include <set>
 #include <vector>
 
@@ -47,7 +49,7 @@ struct Machine {
         decltype(mandatoryFeatures) mandatoryFeatures,
         decltype(sshPublicHostKey) sshPublicHostKey);
 
-    ref<Store> openStore() const;
+    kj::Promise<Result<ref<Store>>> openStore() const;
 };
 
 typedef std::vector<Machine> Machines;

lix/libstore/store-api.cc

@@ -16,8 +16,10 @@
 #include "lix/libstore/worker-protocol.hh"
 #include "lix/libutil/users.hh"
 
+#include <mutex>
 #include <nlohmann/json.hpp>
 #include <regex>
+#include <shared_mutex>
 
 using json = nlohmann::json;
 
@@ -1528,9 +1530,9 @@ static std::string extractConnStr(const std::string &proto, const std::string &c
     return connStr;
 }
 
-ref<Store> openStore(const std::string & uri_,
+kj::Promise<Result<ref<Store>>> openStore(const std::string & uri_,
     const StoreConfig::Params & extraParams)
-{
+try {
     auto params = extraParams;
     try {
         auto parsedUri = parseURL(uri_);
@@ -1548,7 +1550,7 @@ ref<Store> openStore(const std::string & uri_,
                     experimentalFeatureSettings.require(store->config().experimentalFeature());
                     store->init();
                     store->config().warnUnknownSettings();
-                    return ref<Store>(store);
+                    co_return ref<Store>(store);
                 }
             }
         }
@@ -1559,40 +1561,45 @@ ref<Store> openStore(const std::string & uri_,
 
         if (auto store = openFromNonUri(uri, params)) {
             store->config().warnUnknownSettings();
-            return ref<Store>(store);
+            co_return ref<Store>(store);
         }
     }
 
     throw Error("don't know how to open Nix store '%s'", uri_);
+} catch (...) {
+    co_return result::current_exception();
 }
 
 kj::Promise<Result<std::list<ref<Store>>>> getDefaultSubstituters()
 try {
-    static auto stores([]() {
-        std::list<ref<Store>> stores;
+    static std::shared_mutex mtx;
+    static std::optional<std::list<ref<Store>>> stores;
 
+    if (std::shared_lock l(mtx); stores.has_value()) {
+        co_return *stores;
+    }
+
+    std::lock_guard l(mtx);
+
+    if (!stores.has_value()) {
         StringSet done;
 
-        auto addStore = [&](const std::string & uri) {
-            if (!done.insert(uri).second) return;
+        stores.emplace();
+        for (auto uri : settings.substituters.get()) {
+            if (!done.insert(uri).second) continue;
             try {
-                stores.push_back(openStore(uri));
+                stores->push_back(TRY_AWAIT(openStore(uri)));
             } catch (Error & e) {
                 logWarning(e.info());
             }
-        };
+        }
 
-        for (auto uri : settings.substituters.get())
-            addStore(uri);
-
-        stores.sort([](ref<Store> & a, ref<Store> & b) {
+        stores->sort([](ref<Store> & a, ref<Store> & b) {
             return a->config().priority < b->config().priority;
         });
+    }
 
-        return stores;
-    } ());
-
-    co_return stores;
+    co_return *stores;
 } catch (...) {
     co_return result::current_exception();
 }

lix/libstore/store-api.hh

@@ -1000,7 +1000,7 @@ OutputPathMap resolveDerivedPath(Store &, const DerivedPath::Built &, Store * ev
  * You can pass parameters to the store implementation by appending
  * ‘?key=value&key=value&...’ to the URI.
  */
-ref<Store> openStore(const std::string & uri = settings.storeUri.get(),
+kj::Promise<Result<ref<Store>>> openStore(const std::string & uri = settings.storeUri.get(),
     const StoreConfig::Params & extraParams = {});
 
 

lix/nix/daemon.cc

@@ -208,12 +208,14 @@ static PeerInfo getPeerInfo(int remote)
 /**
  * Open a store without a path info cache.
  */
-static ref<Store> openUncachedStore()
-{
+static kj::Promise<Result<ref<Store>>> openUncachedStore()
+try {
     StoreConfig::Params params; // FIXME: get params from somewhere
     // Disable caching since the client already does that.
     params["path-info-cache-size"] = "0";
-    return openStore(settings.storeUri, params);
+    co_return TRY_AWAIT(openStore(settings.storeUri, params));
+} catch (...) {
+    co_return result::current_exception();
 }
 
 /**
@@ -349,7 +351,9 @@ static void daemonLoopImpl(std::optional<TrustedFlag> forceTrustClientOpt)
                 //  Handle the connection.
                 FdSource from(remote.get());
                 FdSink to(remote.get());
-                processConnection(aio, openUncachedStore(), from, to, trusted, NotRecursive);
+                processConnection(
+                    aio, aio.blockOn(openUncachedStore()), from, to, trusted, NotRecursive
+                );
 
                 exit(0);
             }, options).release();
@@ -445,7 +449,7 @@ static void
 runDaemon(AsyncIoRoot & aio, bool stdio, std::optional<TrustedFlag> forceTrustClientOpt)
 {
     if (stdio) {
-        auto store = openUncachedStore();
+        auto store = aio.blockOn(openUncachedStore());
 
         // If --force-untrusted is passed, we cannot forward the connection and
         // must process it ourselves (before delegating to the next store) to

lix/nix/flake.cc

@@ -1096,7 +1096,7 @@ struct CmdFlakeArchive : FlakeCommand, MixJSON, MixDryRun
         }
 
         if (!dryRun && !dstUri.empty()) {
-            ref<Store> dstStore = dstUri.empty() ? openStore() : openStore(dstUri);
+            ref<Store> dstStore = aio().blockOn(dstUri.empty() ? openStore() : openStore(dstUri));
             copyPaths(*store, *dstStore, sources);
         }
     }

lix/nix/main.cc

@@ -260,7 +260,7 @@ static void showHelp(AsyncIoRoot & aio, std::vector<std::string> subcommand, Nix
 
     evalSettings.restrictEval.override(false);
     evalSettings.pureEval.override(false);
-    Evaluator evaluator(aio, {}, openStore("dummy://"));
+    Evaluator evaluator(aio, {}, aio.blockOn(openStore("dummy://")));
     auto state = evaluator.begin(aio);
 
     auto vGenerateManpage = evaluator.mem.allocValue();
@@ -419,7 +419,7 @@ void mainWrapped(AsyncIoRoot & aio, int argc, char * * argv)
             | Xp::FetchClosure
             | Xp::DynamicDerivations);
         evalSettings.pureEval.override(false);
-        Evaluator state(aio, {}, openStore("dummy://"));
+        Evaluator state(aio, {}, aio.blockOn(openStore("dummy://")));
         auto res = nlohmann::json::object();
         res["builtins"] = ({
             auto builtinsJson = nlohmann::json::object();

lix/nix/make-content-addressed.cc

@@ -30,7 +30,7 @@ struct CmdMakeContentAddressed : virtual CopyCommand, virtual StorePathsCommand,
 
     void run(ref<Store> srcStore, StorePaths && storePaths) override
     {
-        auto dstStore = dstUri.empty() ? openStore() : openStore(dstUri);
+        auto dstStore = aio().blockOn(dstUri.empty() ? openStore() : openStore(dstUri));
 
         auto remappings = makeContentAddressed(*srcStore, *dstStore,
             StorePathSet(storePaths.begin(), storePaths.end()));

lix/nix/prefetch.cc

@@ -186,7 +186,7 @@ static int main_nix_prefetch_url(AsyncIoRoot & aio, std::string programName, Str
         if (isOutputARealTerminal(StandardOutputStream::Stderr))
             setLogFormat(LogFormat::bar);
 
-        auto store = openStore();
+        auto store = aio.blockOn(openStore());
         auto evaluator = std::make_unique<Evaluator>(aio, myArgs.searchPath, store);
         auto state = evaluator->begin(aio);
 

lix/nix/repl.cc

@@ -87,7 +87,9 @@ struct CmdRepl : RawInstallablesCommand
             }
             return values;
         };
-        AbstractNixRepl::run(searchPath, openStore(), *state, getValues, {}, getAutoArgs(*evaluator));
+        AbstractNixRepl::run(
+            searchPath, aio().blockOn(openStore()), *state, getValues, {}, getAutoArgs(*evaluator)
+        );
     }
 };
 

lix/nix/sigs.cc

@@ -36,7 +36,7 @@ struct CmdCopySigs : StorePathsCommand
         // FIXME: factor out commonality with MixVerify.
         std::vector<ref<Store>> substituters;
         for (auto & s : substituterUris)
-            substituters.push_back(openStore(s));
+            substituters.push_back(aio().blockOn(openStore(s)));
 
         ThreadPool pool{"CopySigs pool"};
 

lix/nix/verify.cc

@@ -63,7 +63,7 @@ struct CmdVerify : StorePathsCommand
     {
         std::vector<ref<Store>> substituters;
         for (auto & s : substituterUris)
-            substituters.push_back(openStore(s));
+            substituters.push_back(aio().blockOn(openStore(s)));
 
         auto publicKeys = getDefaultPublicKeys();
 

perl/lib/Nix/Store.xs

@@ -26,13 +26,19 @@
 using namespace nix;
 
 
+static AsyncIoRoot & aio()
+{
+    static thread_local AsyncIoRoot root;
+    return root;
+}
+
 static ref<Store> store()
 {
     static std::shared_ptr<Store> _store;
     if (!_store) {
         try {
             initLibStore();
-            _store = openStore();
+            _store = aio().blockOn(openStore());
         } catch (Error & e) {
             croak("%s", e.what());
         }
@@ -40,12 +46,6 @@ static ref<Store> store()
     return ref<Store>(_store);
 }
 
-static AsyncIoRoot & aio()
-{
-    static thread_local AsyncIoRoot root;
-    return root;
-}
-
 
 MODULE = Nix::Store PACKAGE = Nix::Store
 PROTOTYPES: ENABLE

subprojects/nix-eval-jobs/src/nix-eval-jobs.cc

@@ -66,9 +66,9 @@ struct Proc {
                 debug("created worker process %d", getpid());
                 try {
                     AsyncIoRoot aio;
-                    auto evalStore = myArgs.evalStoreUrl
+                    auto evalStore = aio.blockOn(myArgs.evalStoreUrl
                                          ? openStore(*myArgs.evalStoreUrl)
-                                         : openStore();
+                                         : openStore());
                     auto evaluator =
                         nix::make_ref<nix::eval_cache::CachingEvaluator>(
                             aio, myArgs.searchPath, evalStore);

tests/functional/test-libstoreconsumer/main.cc

@@ -19,7 +19,7 @@ int main (int argc, char **argv)
 
         initLibStore();
 
-        auto store = nix::openStore();
+        auto store = aio.blockOn(nix::openStore());
 
         // build the derivation
 

tests/unit/libcmd/args.cc

@@ -32,7 +32,7 @@ TEST(Arguments, lookupFileArg) {
     SearchPath searchPath;
     searchPath.elements.push_back(SearchPath::Elem::parse(searchPathElem));
 
-    auto store = openStore("dummy://");
+    auto store = aio.blockOn(openStore("dummy://"));
     auto state = std::make_shared<Evaluator>(aio, searchPath, store, store);
 
     SourcePath const foundUnitData = aio.blockOn(lookupFileArg(*state, "<example>"));

tests/unit/libexpr-support/tests/libexpr.hh

@@ -31,7 +31,6 @@ namespace nix {
                 , state(*statePtr)
             {
             }
-            ~LibExprTest() noexcept = default;
             Value eval(std::string input, bool forceValue = true, const FeatureSettings & fSettings = featureSettings) {
                 Value v;
                 Expr & e = evaluator.parseExprFromString(input, CanonPath::root, fSettings);
@@ -45,7 +44,6 @@ namespace nix {
                 return evaluator.symbols.create(value);
             }
 
-            AsyncIoRoot aio;
             Evaluator evaluator;
             box_ptr<EvalState> statePtr;
             EvalState & state;

tests/unit/libstore-support/tests/libstore.hh

@@ -5,6 +5,7 @@
 #include <gmock/gmock.h>
 
 #include "lix/libstore/store-api.hh"
+#include "lix/libutil/async.hh"
 
 namespace nix {
 
@@ -16,9 +17,12 @@ class LibStoreTest : public ::testing::Test {
 
     protected:
         LibStoreTest()
-            : store(openStore("dummy://"))
+            : store(aio.blockOn(openStore("dummy://")))
         { }
 
+        ~LibStoreTest() noexcept(true) {}
+
+        AsyncIoRoot aio;
         ref<Store> store;
 };