Add paths to the store asynchronously

Adding paths to the store can be slow due to I/O overhead, but
especially when going through the daemon because of the round-trip
latency of every wopAddToStore call.

So we now do the addToStore() calls asynchronously from a separate
thread from the evaluator. This slightly speeds up the local store,
and makes going through the daemon almost as fast as a local store.

src/libcmd/installable-attr-path.cc

@@ -89,7 +89,8 @@ DerivedPathsWithInfo InstallableAttrPath::toDerivedPaths()
     }
 
     DerivedPathsWithInfo res;
-    for (auto & [drvPath, outputs] : byDrvPath)
+    for (auto & [drvPath, outputs] : byDrvPath) {
+        state->waitForPath(drvPath);
         res.push_back({
             .path =
                 DerivedPath::Built{
@@ -102,6 +103,7 @@ DerivedPathsWithInfo InstallableAttrPath::toDerivedPaths()
                    so we can fill in this info. */
             }),
         });
+    }
 
     return res;
 }

src/libcmd/installable-flake.cc

@@ -102,6 +102,7 @@ DerivedPathsWithInfo InstallableFlake::toDerivedPaths()
     }
 
     auto drvPath = attr->forceDerivation();
+    state->waitForPath(drvPath);
 
     std::optional<NixInt::Inner> priority;
 

src/libcmd/repl.cc

@@ -333,6 +333,7 @@ StorePath NixRepl::getDerivationPath(Value & v)
     auto drvPath = packageInfo->queryDrvPath();
     if (!drvPath)
         throw Error("expression did not evaluate to a valid derivation (no 'drvPath' attribute)");
+    state->waitForPath(*drvPath);
     if (!state->store->isValidPath(*drvPath))
         throw Error("expression evaluated to invalid derivation '%s'", state->store->printStorePath(*drvPath));
     return *drvPath;

src/libexpr-c/nix_api_expr.cc

@@ -69,6 +69,7 @@ nix_err nix_expr_eval_from_string(
         nix::Expr * parsedExpr = state->state.parseExprFromString(expr, state->state.rootPath(nix::CanonPath(path)));
         state->state.eval(parsedExpr, value->value);
         state->state.forceValue(value->value, nix::noPos);
+        state->state.waitForAllPaths();
     }
     NIXC_CATCH_ERRS
 }
@@ -80,6 +81,7 @@ nix_err nix_value_call(nix_c_context * context, EvalState * state, Value * fn, n
     try {
         state->state.callFunction(fn->value, arg->value, value->value, nix::noPos);
         state->state.forceValue(value->value, nix::noPos);
+        state->state.waitForAllPaths();
     }
     NIXC_CATCH_ERRS
 }
@@ -92,6 +94,7 @@ nix_err nix_value_call_multi(
     try {
         state->state.callFunction(fn->value, {(nix::Value **) args, nargs}, value->value, nix::noPos);
         state->state.forceValue(value->value, nix::noPos);
+        state->state.waitForAllPaths();
     }
     NIXC_CATCH_ERRS
 }
@@ -102,6 +105,7 @@ nix_err nix_value_force(nix_c_context * context, EvalState * state, nix_value *
         context->last_err_code = NIX_OK;
     try {
         state->state.forceValue(value->value, nix::noPos);
+        state->state.waitForAllPaths();
     }
     NIXC_CATCH_ERRS
 }
@@ -112,6 +116,7 @@ nix_err nix_value_force_deep(nix_c_context * context, EvalState * state, nix_val
         context->last_err_code = NIX_OK;
     try {
         state->state.forceValueDeep(value->value);
+        state->state.waitForAllPaths();
     }
     NIXC_CATCH_ERRS
 }

src/libexpr-c/nix_api_value.cc

@@ -345,6 +345,7 @@ nix_value * nix_get_attr_byname(nix_c_context * context, const nix_value * value
         if (attr) {
             nix_gc_incref(nullptr, attr->value);
             state->state.forceValue(*attr->value, nix::noPos);
+            state->state.waitForAllPaths();
             return as_nix_value_ptr(attr->value);
         }
         nix_set_err_msg(context, NIX_ERR_KEY, "missing attribute");

src/libexpr/eval-cache.cc

@@ -708,6 +708,7 @@ StorePath AttrCursor::forceDerivation()
         /* The eval cache contains 'drvPath', but the actual path has
            been garbage-collected. So force it to be regenerated. */
         aDrvPath->forceValue();
+        root->state.waitForPath(drvPath);
         if (!root->state.store->isValidPath(drvPath))
             throw Error(
                 "don't know how to recreate store derivation '%s'!", root->state.store->printStorePath(drvPath));

src/libexpr/eval.cc

@@ -21,6 +21,7 @@
 #include "nix/fetchers/fetch-to-store.hh"
 #include "nix/fetchers/tarball.hh"
 #include "nix/fetchers/input-cache.hh"
+#include "nix/store/async-path-writer.hh"
 
 #include "parser-tab.hh"
 
@@ -326,6 +327,7 @@ EvalState::EvalState(
     , debugRepl(nullptr)
     , debugStop(false)
     , trylevel(0)
+    , asyncPathWriter(AsyncPathWriter::make(store))
     , regexCache(makeRegexCache())
 #if NIX_USE_BOEHMGC
     , valueAllocCache(std::allocate_shared<void *>(traceable_allocator<void *>(), nullptr))
@@ -1024,6 +1026,7 @@ std::string EvalState::mkSingleDerivedPathStringRaw(const SingleDerivedPath & p)
                 auto optStaticOutputPath = std::visit(
                     overloaded{
                         [&](const SingleDerivedPath::Opaque & o) {
+                            waitForPath(o.path);
                             auto drv = store->readDerivation(o.path);
                             auto i = drv.outputs.find(b.output);
                             if (i == drv.outputs.end())
@@ -3249,4 +3252,24 @@ void forceNoNullByte(std::string_view s, std::function<Pos()> pos)
     }
 }
 
+void EvalState::waitForPath(const StorePath & path)
+{
+    asyncPathWriter->waitForPath(path);
+}
+
+void EvalState::waitForPath(const SingleDerivedPath & path)
+{
+    std::visit(
+        overloaded{
+            [&](const DerivedPathOpaque & p) { waitForPath(p.path); },
+            [&](const SingleDerivedPathBuilt & p) { waitForPath(*p.drvPath); },
+        },
+        path.raw());
+}
+
+void EvalState::waitForAllPaths()
+{
+    asyncPathWriter->waitForAllPaths();
+}
+
 } // namespace nix

src/libexpr/include/nix/expr/eval.hh

@@ -45,6 +45,7 @@ class StorePath;
 struct SingleDerivedPath;
 enum RepairFlag : bool;
 struct MemorySourceAccessor;
+struct AsyncPathWriter;
 
 namespace eval_cache {
 class EvalCache;
@@ -320,6 +321,8 @@ public:
     std::list<DebugTrace> debugTraces;
     std::map<const Expr *, const std::shared_ptr<const StaticEnv>> exprEnvs;
 
+    ref<AsyncPathWriter> asyncPathWriter;
+
     const std::shared_ptr<const StaticEnv> getStaticEnv(const Expr & expr) const
     {
         auto i = exprEnvs.find(&expr);
@@ -907,6 +910,10 @@ public:
 
     DocComment getDocCommentForPos(PosIdx pos);
 
+    void waitForPath(const StorePath & path);
+    void waitForPath(const SingleDerivedPath & path);
+    void waitForAllPaths();
+
 private:
 
     /**

src/libexpr/primops.cc

@@ -63,6 +63,7 @@ StringMap EvalState::realiseContext(const NixStringContext & context, StorePathS
 
     for (auto & c : context) {
         auto ensureValid = [&](const StorePath & p) {
+            waitForPath(p);
             if (!store->isValidPath(p))
                 error<InvalidPathError>(store->printStorePath(p)).debugThrow();
         };
@@ -291,6 +292,7 @@ static void import(EvalState & state, const PosIdx pos, Value & vPath, Value * v
         if (!state.store->isStorePath(path2))
             return std::nullopt;
         auto storePath = state.store->parseStorePath(path2);
+        state.waitForPath(storePath);
         if (!(state.store->isValidPath(storePath) && isDerivation(path2)))
             return std::nullopt;
         return storePath;
@@ -1583,6 +1585,8 @@ static void derivationStrictInternal(EvalState & state, std::string_view drvName
                 [&](const NixStringContextElem::DrvDeep & d) {
                     /* !!! This doesn't work if readOnlyMode is set. */
                     StorePathSet refs;
+                    // FIXME: don't need to wait, we only need the references.
+                    state.waitForPath(d.drvPath);
                     state.store->computeFSClosure(d.drvPath, refs);
                     for (auto & j : refs) {
                         drv.inputSrcs.insert(j);
@@ -1707,7 +1711,7 @@ static void derivationStrictInternal(EvalState & state, std::string_view drvName
     }
 
     /* Write the resulting term into the Nix store directory. */
-    auto drvPath = writeDerivation(*state.store, drv, state.repair);
+    auto drvPath = writeDerivation(*state.store, *state.asyncPathWriter, drv, state.repair);
     auto drvPathS = state.store->printStorePath(drvPath);
 
     printMsg(lvlChatty, "instantiated '%1%' -> '%2%'", drvName, drvPathS);

src/libexpr/primops/context.cc

@@ -61,6 +61,7 @@ static void prim_unsafeDiscardOutputDependency(EvalState & state, const PosIdx p
     NixStringContext context2;
     for (auto && c : context) {
         if (auto * ptr = std::get_if<NixStringContextElem::DrvDeep>(&c.raw)) {
+            state.waitForPath(ptr->drvPath); // FIXME: why?
             context2.emplace(NixStringContextElem::Opaque{.path = ptr->drvPath});
         } else {
             /* Can reuse original item */

src/libstore/async-path-writer.cc

@@ -0,0 +1,151 @@
+#include "nix/store/async-path-writer.hh"
+#include "nix/util/archive.hh"
+
+#include <thread>
+#include <future>
+
+namespace nix {
+
+struct AsyncPathWriterImpl : AsyncPathWriter
+{
+    ref<Store> store;
+
+    struct Item
+    {
+        StorePath storePath;
+        std::string contents;
+        std::string name;
+        Hash hash;
+        StorePathSet references;
+        RepairFlag repair;
+        std::promise<void> promise;
+    };
+
+    struct State
+    {
+        std::vector<Item> items;
+        std::unordered_map<StorePath, std::shared_future<void>> futures;
+        bool quit = false;
+    };
+
+    Sync<State> state_;
+
+    std::thread workerThread;
+
+    std::condition_variable wakeupCV;
+
+    AsyncPathWriterImpl(ref<Store> store)
+        : store(store)
+    {
+        workerThread = std::thread([&]() {
+            while (true) {
+                std::vector<Item> items;
+
+                {
+                    auto state(state_.lock());
+                    while (!state->quit && state->items.empty())
+                        state.wait(wakeupCV);
+                    if (state->items.empty() && state->quit)
+                        return;
+                    std::swap(items, state->items);
+                }
+
+                try {
+                    writePaths(items);
+                    for (auto & item : items)
+                        item.promise.set_value();
+                } catch (...) {
+                    for (auto & item : items)
+                        item.promise.set_exception(std::current_exception());
+                }
+            }
+        });
+    }
+
+    ~AsyncPathWriterImpl()
+    {
+        state_.lock()->quit = true;
+        wakeupCV.notify_all();
+        workerThread.join();
+    }
+
+    StorePath
+    addPath(std::string contents, std::string name, StorePathSet references, RepairFlag repair, bool readOnly) override
+    {
+        auto hash = hashString(HashAlgorithm::SHA256, contents);
+
+        auto storePath = store->makeFixedOutputPathFromCA(
+            name,
+            TextInfo{
+                .hash = hash,
+                .references = references,
+            });
+
+        if (!readOnly) {
+            auto state(state_.lock());
+            std::promise<void> promise;
+            state->futures.insert_or_assign(storePath, promise.get_future());
+            state->items.push_back(
+                Item{
+                    .storePath = storePath,
+                    .contents = std::move(contents),
+                    .name = std::move(name),
+                    .hash = hash,
+                    .references = std::move(references),
+                    .repair = repair,
+                    .promise = std::move(promise),
+                });
+            wakeupCV.notify_all();
+        }
+
+        return storePath;
+    }
+
+    void waitForPath(const StorePath & path) override
+    {
+        auto future = ({
+            auto state = state_.lock();
+            auto i = state->futures.find(path);
+            if (i == state->futures.end())
+                return;
+            i->second;
+        });
+        future.get();
+    }
+
+    void waitForAllPaths() override
+    {
+        auto futures = ({
+            auto state(state_.lock());
+            std::move(state->futures);
+        });
+        for (auto & future : futures)
+            future.second.get();
+    }
+
+    void writePaths(const std::vector<Item> & items)
+    {
+        // FIXME: use addMultipeToStore() once it doesn't require a
+        // NAR hash from the client for CA objects.
+
+        for (auto & item : items) {
+            StringSource source(item.contents);
+            auto storePath = store->addToStoreFromDump(
+                source,
+                item.storePath.name(),
+                FileSerialisationMethod::Flat,
+                ContentAddressMethod::Raw::Text,
+                HashAlgorithm::SHA256,
+                item.references,
+                item.repair);
+            assert(storePath == item.storePath);
+        }
+    }
+};
+
+ref<AsyncPathWriter> AsyncPathWriter::make(ref<Store> store)
+{
+    return make_ref<AsyncPathWriterImpl>(store);
+}
+
+} // namespace nix

src/libstore/derivations.cc

@@ -9,6 +9,7 @@
 #include "nix/store/common-protocol-impl.hh"
 #include "nix/util/strings-inline.hh"
 #include "nix/util/json-utils.hh"
+#include "nix/store/async-path-writer.hh"
 
 #include <boost/container/small_vector.hpp>
 #include <nlohmann/json.hpp>
@@ -133,6 +134,20 @@ StorePath writeDerivation(Store & store, const Derivation & drv, RepairFlag repa
                                                });
 }
 
+StorePath writeDerivation(
+    Store & store, AsyncPathWriter & asyncPathWriter, const Derivation & drv, RepairFlag repair, bool readOnly)
+{
+    auto references = drv.inputSrcs;
+    for (auto & i : drv.inputDrvs.map)
+        references.insert(i.first);
+    return asyncPathWriter.addPath(
+        drv.unparse(store, false),
+        std::string(drv.name) + drvExtension,
+        references,
+        repair,
+        readOnly || settings.readOnlyMode);
+}
+
 namespace {
 /**
  * This mimics std::istream to some extent. We use this much smaller implementation

src/libstore/include/nix/store/async-path-writer.hh

@@ -0,0 +1,19 @@
+#pragma once
+
+#include "nix/store/store-api.hh"
+
+namespace nix {
+
+struct AsyncPathWriter
+{
+    virtual StorePath addPath(
+        std::string contents, std::string name, StorePathSet references, RepairFlag repair, bool readOnly = false) = 0;
+
+    virtual void waitForPath(const StorePath & path) = 0;
+
+    virtual void waitForAllPaths() = 0;
+
+    static ref<AsyncPathWriter> make(ref<Store> store);
+};
+
+} // namespace nix

src/libstore/include/nix/store/derivations.hh

@@ -17,6 +17,7 @@
 namespace nix {
 
 struct StoreDirConfig;
+struct AsyncPathWriter;
 
 /* Abstract syntax of derivations. */
 
@@ -412,6 +413,16 @@ class Store;
  */
 StorePath writeDerivation(Store & store, const Derivation & drv, RepairFlag repair = NoRepair, bool readOnly = false);
 
+/**
+ * Asynchronously write a derivation to the Nix store, and return its path.
+ */
+StorePath writeDerivation(
+    Store & store,
+    AsyncPathWriter & asyncPathWriter,
+    const Derivation & drv,
+    RepairFlag repair = NoRepair,
+    bool readOnly = false);
+
 /**
  * Read a derivation from a file.
  */

src/libstore/include/nix/store/meson.build

@@ -10,6 +10,7 @@ config_pub_h = configure_file(
 )
 
 headers = [ config_pub_h ] + files(
+  'async-path-writer.hh',
   'binary-cache-store.hh',
   'build-result.hh',
   'build/derivation-building-goal.hh',

src/libstore/meson.build

@@ -262,6 +262,7 @@ config_priv_h = configure_file(
 subdir('nix-meson-build-support/common')
 
 sources = files(
+  'async-path-writer.cc',
   'binary-cache-store.cc',
   'build-result.cc',
   'build/derivation-building-goal.cc',

src/nix/app.cc

@@ -74,6 +74,7 @@ UnresolvedApp InstallableValue::toApp(EvalState & state)
                 std::visit(
                     overloaded{
                         [&](const NixStringContextElem::DrvDeep & d) -> DerivedPath {
+                            state.waitForPath(d.drvPath);
                             /* We want all outputs of the drv */
                             return DerivedPath::Built{
                                 .drvPath = makeConstantStorePathRef(d.drvPath),
@@ -81,6 +82,7 @@ UnresolvedApp InstallableValue::toApp(EvalState & state)
                             };
                         },
                         [&](const NixStringContextElem::Built & b) -> DerivedPath {
+                            state.waitForPath(*b.drvPath);
                             return DerivedPath::Built{
                                 .drvPath = b.drvPath,
                                 .outputs = OutputsSpec::Names{b.output},

src/nix/nix-build/nix-build.cc

@@ -451,7 +451,9 @@ static void main_nix_build(int argc, char ** argv)
             throw UsageError("nix-shell requires a single derivation");
 
         auto & packageInfo = drvs.front();
-        auto drv = evalStore->derivationFromPath(packageInfo.requireDrvPath());
+        auto drvPath = packageInfo.requireDrvPath();
+        state->waitForPath(drvPath);
+        auto drv = evalStore->derivationFromPath(drvPath);
 
         std::vector<DerivedPath> pathsToBuild;
         RealisedPath::Set pathsToCopy;
@@ -475,6 +477,7 @@ static void main_nix_build(int argc, char ** argv)
                     throw Error("the 'bashInteractive' attribute in <nixpkgs> did not evaluate to a derivation");
 
                 auto bashDrv = drv->requireDrvPath();
+                state->waitForPath(bashDrv);
                 pathsToBuild.push_back(
                     DerivedPath::Built{
                         .drvPath = makeConstantStorePathRef(bashDrv),
@@ -683,6 +686,7 @@ static void main_nix_build(int argc, char ** argv)
 
         for (auto & packageInfo : drvs) {
             auto drvPath = packageInfo.requireDrvPath();
+            state->waitForPath(drvPath);
 
             auto outputName = packageInfo.queryOutputName();
             if (outputName == "")

src/nix/nix-env/nix-env.cc

@@ -746,6 +746,8 @@ static void opSet(Globals & globals, Strings opFlags, Strings opArgs)
         drv.setName(globals.forceName);
 
     auto drvPath = drv.queryDrvPath();
+    if (drvPath)
+        globals.state->waitForPath(*drvPath);
     std::vector<DerivedPath> paths{
         drvPath ? (DerivedPath) (DerivedPath::Built{
                       .drvPath = makeConstantStorePathRef(*drvPath),

src/nix/nix-env/user-env.cc

@@ -37,8 +37,10 @@ bool createUserEnv(
        exist already. */
     std::vector<StorePathWithOutputs> drvsToBuild;
     for (auto & i : elems)
-        if (auto drvPath = i.queryDrvPath())
+        if (auto drvPath = i.queryDrvPath()) {
+            state.waitForPath(*drvPath);
             drvsToBuild.push_back({*drvPath});
+        }
 
     debug("building user environment dependencies");
     state.store->buildPaths(toDerivedPaths(drvsToBuild), state.repair ? bmRepair : bmNormal);
@@ -151,6 +153,7 @@ bool createUserEnv(
     debug("building user environment");
     std::vector<StorePathWithOutputs> topLevelDrvs;
     topLevelDrvs.push_back({topLevelDrv});
+    state.waitForPath(topLevelDrv);
     state.store->buildPaths(toDerivedPaths(topLevelDrvs), state.repair ? bmRepair : bmNormal);
 
     /* Switch the current user environment to the output path. */

tests/functional/flakes/eval-cache.sh

@@ -48,3 +48,11 @@ nix build --no-link "$flake1Dir#stack-depth"
 expect 1 nix build "$flake1Dir#ifd" --option allow-import-from-derivation false 2>&1 \
   | grepQuiet 'error: cannot build .* during evaluation because the option '\''allow-import-from-derivation'\'' is disabled'
 nix build --no-link "$flake1Dir#ifd"
+
+# Test that a store derivation is recreated when it has been deleted
+# but the corresponding attribute is still cached.
+if ! isTestOnNixOS; then
+    nix build "$flake1Dir#drv"
+    clearStore
+    nix build "$flake1Dir#drv"
+fi