From 14067da9472b6ea5acf28364df64cc4112bd2fb3 Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Mon, 3 Mar 2025 20:48:59 +0100 Subject: [PATCH] libstore: pass async streams to Store::add{,Multiple}ToStore Change-Id: Idafc0d640bf66d2d28ff71ae81546db8cf463ee2 --- lix/legacy/nix-store.cc | 4 +- lix/libcmd/cmd-profiles.cc | 3 +- lix/libfetchers/tarball.cc | 3 +- lix/libstore/binary-cache-store.cc | 13 ++++--- lix/libstore/binary-cache-store.hh | 2 +- lix/libstore/build/local-derivation-goal.cc | 2 +- lix/libstore/daemon.cc | 9 +++-- lix/libstore/dummy-store.cc | 2 +- lix/libstore/export-import.cc | 2 +- lix/libstore/legacy-ssh-store.cc | 6 +-- lix/libstore/local-store.cc | 17 ++++++--- lix/libstore/local-store.hh | 2 +- lix/libstore/make-content-addressed.cc | 3 +- lix/libstore/remote-store.cc | 28 ++++++++++---- lix/libstore/remote-store.hh | 2 +- lix/libstore/store-api.cc | 41 +++++++++++++-------- lix/libstore/store-api.hh | 10 +++-- lix/nix/add-to-store.cc | 3 +- 18 files changed, 98 insertions(+), 54 deletions(-) diff --git a/lix/legacy/nix-store.cc b/lix/legacy/nix-store.cc index a93e9899b..00a23cbe8 100644 --- a/lix/legacy/nix-store.cc +++ b/lix/legacy/nix-store.cc @@ -1,6 +1,7 @@ #include "lix/libutil/archive.hh" #include "lix/libstore/derivations.hh" #include "dotgraph.hh" +#include "lix/libutil/async-io.hh" #include "lix/libutil/async.hh" #include "lix/libutil/exit.hh" #include "lix/libstore/globals.hh" @@ -1015,8 +1016,9 @@ static void opServe(AsyncIoRoot & aio, Strings opFlags, Strings opArgs) throw Error("narInfo is too old and missing the narSize field"); SizedSource sizedSource(in, info.narSize); + AsyncSourceInputStream stream{sizedSource}; - aio.blockOn(store->addToStore(info, sizedSource, NoRepair, NoCheckSigs)); + aio.blockOn(store->addToStore(info, stream, NoRepair, NoCheckSigs)); // consume all the data that has been sent before continuing. sizedSource.drainAll(); diff --git a/lix/libcmd/cmd-profiles.cc b/lix/libcmd/cmd-profiles.cc index 438556847..f16d8de93 100644 --- a/lix/libcmd/cmd-profiles.cc +++ b/lix/libcmd/cmd-profiles.cc @@ -3,6 +3,7 @@ #include "lix/libcmd/cmd-profiles.hh" #include "lix/libcmd/built-path.hh" #include "lix/libstore/builtins/buildenv.hh" +#include "lix/libutil/async-io.hh" #include "lix/libutil/logging.hh" #include "lix/libstore/names.hh" #include "lix/libstore/store-api.hh" @@ -266,7 +267,7 @@ try { }; info.narSize = sink.s.size(); - StringSource source(sink.s); + AsyncStringInputStream source(sink.s); TRY_AWAIT(store->addToStore(info, source)); co_return std::move(info.path); diff --git a/lix/libfetchers/tarball.cc b/lix/libfetchers/tarball.cc index fe9f331c1..51faace1b 100644 --- a/lix/libfetchers/tarball.cc +++ b/lix/libfetchers/tarball.cc @@ -5,6 +5,7 @@ #include "lix/libfetchers/builtin-fetchers.hh" #include "lix/libstore/store-api.hh" #include "lix/libutil/archive.hh" +#include "lix/libutil/async-io.hh" #include "lix/libutil/async.hh" #include "lix/libutil/tarfile.hh" #include "lix/libstore/temporary-dir.hh" @@ -88,7 +89,7 @@ try { hashString(HashType::SHA256, sink.s), }; info.narSize = sink.s.size(); - auto source = StringSource { sink.s }; + auto source = AsyncStringInputStream { sink.s }; TRY_AWAIT(store->addToStore(info, source, NoRepair, NoCheckSigs)); storePath = std::move(info.path); } diff --git a/lix/libstore/binary-cache-store.cc b/lix/libstore/binary-cache-store.cc index efdb4eab1..c16ce970a 100644 --- a/lix/libstore/binary-cache-store.cc +++ b/lix/libstore/binary-cache-store.cc @@ -267,17 +267,20 @@ try { co_return result::current_exception(); } -kj::Promise> BinaryCacheStore::addToStore(const ValidPathInfo & info, Source & narSource, - RepairFlag repair, CheckSigsFlag checkSigs) +kj::Promise> BinaryCacheStore::addToStore( + const ValidPathInfo & info, + AsyncInputStream & narSource, + RepairFlag repair, + CheckSigsFlag checkSigs +) try { if (!repair && isValidPath(info.path)) { // FIXME: copyNAR -> null sink - narSource.drain(); + TRY_AWAIT(narSource.drain()); co_return result::success(); } - AsyncSourceInputStream stream{narSource}; - TRY_AWAIT(addToStoreCommon(stream, repair, checkSigs, {[&](HashResult nar) { + TRY_AWAIT(addToStoreCommon(narSource, repair, checkSigs, {[&](HashResult nar) { /* FIXME reinstate these, once we can correctly do hash modulo sink as needed. We need to throw here in case we uploaded a corrupted store path. */ // assert(info.narHash == nar.first); diff --git a/lix/libstore/binary-cache-store.hh b/lix/libstore/binary-cache-store.hh index 490106fa5..efa0bbdd0 100644 --- a/lix/libstore/binary-cache-store.hh +++ b/lix/libstore/binary-cache-store.hh @@ -115,7 +115,7 @@ public: std::optional queryPathFromHashPart(const std::string & hashPart) override; - kj::Promise> addToStore(const ValidPathInfo & info, Source & narSource, + kj::Promise> addToStore(const ValidPathInfo & info, AsyncInputStream & narSource, RepairFlag repair, CheckSigsFlag checkSigs) override; kj::Promise> addToStoreFromDump( diff --git a/lix/libstore/build/local-derivation-goal.cc b/lix/libstore/build/local-derivation-goal.cc index da1c4b50a..1b7c5fc15 100644 --- a/lix/libstore/build/local-derivation-goal.cc +++ b/lix/libstore/build/local-derivation-goal.cc @@ -1096,7 +1096,7 @@ struct RestrictedStore : public virtual IndirectRootStore, public virtual GcStor RepairFlag repair) override try { throw Error("addToStoreFlat"); } catch (...) { return {result::current_exception()}; } - kj::Promise> addToStore(const ValidPathInfo & info, Source & narSource, + kj::Promise> addToStore(const ValidPathInfo & info, AsyncInputStream & narSource, RepairFlag repair = NoRepair, CheckSigsFlag checkSigs = CheckSigs) override try { TRY_AWAIT(next->addToStore(info, narSource, repair, checkSigs)); diff --git a/lix/libstore/daemon.cc b/lix/libstore/daemon.cc index 93ae21696..fcc8b9404 100644 --- a/lix/libstore/daemon.cc +++ b/lix/libstore/daemon.cc @@ -524,8 +524,9 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref store *store, WorkerProto::ReadConn{source, clientVersion} ); info.ultimate = false; // duplicated in RemoteStore::addMultipleToStore + AsyncSourceInputStream stream{source}; aio.blockOn(store->addToStore( - info, source, RepairFlag{repair}, dontCheckSigs ? NoCheckSigs : CheckSigs + info, stream, RepairFlag{repair}, dontCheckSigs ? NoCheckSigs : CheckSigs )); } } @@ -910,7 +911,8 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref store logger->startWork(); { FramedSource source(from); - aio.blockOn(store->addToStore(info, source, (RepairFlag) repair, + AsyncSourceInputStream stream{source}; + aio.blockOn(store->addToStore(info, stream, (RepairFlag) repair, dontCheckSigs ? NoCheckSigs : CheckSigs)); } logger->stopWork(); @@ -923,7 +925,8 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref store logger->startWork(); // FIXME: race if addToStore doesn't read source? - aio.blockOn(store->addToStore(info, *source, (RepairFlag) repair, + AsyncSourceInputStream stream{*source}; + aio.blockOn(store->addToStore(info, stream, (RepairFlag) repair, dontCheckSigs ? NoCheckSigs : CheckSigs)); logger->stopWork(); diff --git a/lix/libstore/dummy-store.cc b/lix/libstore/dummy-store.cc index cbadac2eb..3503dee2b 100644 --- a/lix/libstore/dummy-store.cc +++ b/lix/libstore/dummy-store.cc @@ -54,7 +54,7 @@ struct DummyStore final : public Store std::optional queryPathFromHashPart(const std::string & hashPart) override { unsupported("queryPathFromHashPart"); } - kj::Promise> addToStore(const ValidPathInfo & info, Source & source, + kj::Promise> addToStore(const ValidPathInfo & info, AsyncInputStream & source, RepairFlag repair, CheckSigsFlag checkSigs) override try { unsupported("addToStore"); } catch (...) { return {result::current_exception()}; } diff --git a/lix/libstore/export-import.cc b/lix/libstore/export-import.cc index d8fcc39fc..40d9017f3 100644 --- a/lix/libstore/export-import.cc +++ b/lix/libstore/export-import.cc @@ -92,7 +92,7 @@ try { readString(source); // Can't use underlying source, which would have been exhausted - auto source = StringSource(saved.s); + auto source = AsyncStringInputStream(saved.s); TRY_AWAIT(addToStore(info, source, NoRepair, checkSigs)); res.push_back(info.path); diff --git a/lix/libstore/legacy-ssh-store.cc b/lix/libstore/legacy-ssh-store.cc index a990c17dd..5d97a7185 100644 --- a/lix/libstore/legacy-ssh-store.cc +++ b/lix/libstore/legacy-ssh-store.cc @@ -189,7 +189,7 @@ struct LegacySSHStore final : public Store return info; } - kj::Promise> addToStore(const ValidPathInfo & info, Source & source, + kj::Promise> addToStore(const ValidPathInfo & info, AsyncInputStream & source, RepairFlag repair, CheckSigsFlag checkSigs) override try { debug("adding path '%s' to remote host '%s'", printStorePath(info.path), host); @@ -211,7 +211,7 @@ struct LegacySSHStore final : public Store << info.sigs << renderContentAddress(info.ca); try { - conn->to << copyNAR(source); + TRY_AWAIT(copyNAR(source)->drainInto(conn->to)); } catch (...) { conn->good = false; throw; @@ -224,7 +224,7 @@ struct LegacySSHStore final : public Store << ServeProto::Command::ImportPaths << 1; try { - conn->to << copyNAR(source); + TRY_AWAIT(copyNAR(source)->drainInto(conn->to)); } catch (...) { conn->good = false; throw; diff --git a/lix/libstore/local-store.cc b/lix/libstore/local-store.cc index 471e0187e..7d6044d43 100644 --- a/lix/libstore/local-store.cc +++ b/lix/libstore/local-store.cc @@ -10,6 +10,7 @@ #include "lix/libutil/async.hh" #include "lix/libutil/references.hh" #include "lix/libutil/result.hh" +#include "lix/libutil/serialise.hh" #include "lix/libutil/topo-sort.hh" #include "lix/libutil/signals.hh" #include "lix/libutil/finally.hh" @@ -1207,8 +1208,12 @@ bool LocalStore::realisationIsUntrusted(const Realisation & realisation) return config_.requireSigs && !realisation.checkSignatures(getPublicKeys()); } -kj::Promise> LocalStore::addToStore(const ValidPathInfo & info, Source & source, - RepairFlag repair, CheckSigsFlag checkSigs) +kj::Promise> LocalStore::addToStore( + const ValidPathInfo & info, + AsyncInputStream & source, + RepairFlag repair, + CheckSigsFlag checkSigs +) try { if (checkSigs && pathInfoIsUntrusted(info)) throw Error("cannot add path '%s' because it lacks a signature by a trusted key", printStorePath(info.path)); @@ -1238,10 +1243,10 @@ try { of the NAR. */ HashSink hashSink(HashType::SHA256); - TeeSource wrapperSource { source, hashSink }; + AsyncTeeInputStream wrapperSource { source, hashSink }; narRead = true; - restorePath(realPath, wrapperSource); + TRY_AWAIT(restorePath(realPath, wrapperSource)); auto hashResult = hashSink.finish(); @@ -1279,8 +1284,8 @@ try { } if (!narRead) { - auto copy = copyNAR(source); - while (copy.next()) {} + NullSink null; + TRY_AWAIT(copyNAR(source)->drainInto(null)); } co_return result::success(); } catch (...) { diff --git a/lix/libstore/local-store.hh b/lix/libstore/local-store.hh index 1da2a0416..2e32f82cd 100644 --- a/lix/libstore/local-store.hh +++ b/lix/libstore/local-store.hh @@ -208,7 +208,7 @@ public: bool pathInfoIsUntrusted(const ValidPathInfo &) override; bool realisationIsUntrusted(const Realisation & ) override; - kj::Promise> addToStore(const ValidPathInfo & info, Source & source, + kj::Promise> addToStore(const ValidPathInfo & info, AsyncInputStream & source, RepairFlag repair, CheckSigsFlag checkSigs) override; kj::Promise> addToStoreFromDump( diff --git a/lix/libstore/make-content-addressed.cc b/lix/libstore/make-content-addressed.cc index c4e2c8a88..dd21a51cc 100644 --- a/lix/libstore/make-content-addressed.cc +++ b/lix/libstore/make-content-addressed.cc @@ -1,4 +1,5 @@ #include "lix/libstore/make-content-addressed.hh" +#include "lix/libutil/async-io.hh" #include "lix/libutil/async.hh" #include "lix/libutil/references.hh" #include "lix/libutil/strings.hh" @@ -68,7 +69,7 @@ try { info.narHash = hashString(HashType::SHA256, rewritten); info.narSize = sink.s.size(); - StringSource source(rewritten); + AsyncStringInputStream source(rewritten); TRY_AWAIT(dstStore.addToStore(info, source)); remappings.insert_or_assign(std::move(path), std::move(info.path)); diff --git a/lix/libstore/remote-store.cc b/lix/libstore/remote-store.cc index 5c37f40d3..327311541 100644 --- a/lix/libstore/remote-store.cc +++ b/lix/libstore/remote-store.cc @@ -503,8 +503,12 @@ try { } -kj::Promise> RemoteStore::addToStore(const ValidPathInfo & info, Source & source, - RepairFlag repair, CheckSigsFlag checkSigs) +kj::Promise> RemoteStore::addToStore( + const ValidPathInfo & info, + AsyncInputStream & source, + RepairFlag repair, + CheckSigsFlag checkSigs +) try { auto conn(getConnection()); @@ -518,11 +522,21 @@ try { << repair << !checkSigs; if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 23) { - conn.withFramedSink([&](Sink & sink) { - sink << copyNAR(source); - }); + auto copier = copyNAR(source); + TRY_AWAIT(conn.withFramedSinkAsync([&](Sink & sink) { + return copier->drainInto(sink); + })); } else { - conn.processStderr(0, &source); + IndirectAsyncInputStreamToSource is(source); + auto pfp = kj::newPromiseAndCrossThreadFulfiller(); + auto thread = std::async(std::launch::async, [&] { + KJ_DEFER(pfp.fulfiller->fulfill()); + conn.processStderr(0, &is); + }); + co_await pfp.promise.exclusiveJoin(is.feed()); + // if the thread stops we're always clear. if the feeder stops early (or + // fails) it'll have thrown an exception, and the thread will stop soon. + thread.get(); } co_return result::success(); } catch (...) { @@ -552,7 +566,7 @@ try { sink << WorkerProto::Serialise::write(*this, WorkerProto::WriteConn {remoteVersion}, pathInfo); - TRY_AWAIT(pathSource())->drainInto(sink); + TRY_AWAIT(TRY_AWAIT(pathSource())->drainInto(sink)); } co_return result::success(); } catch (...) { diff --git a/lix/libstore/remote-store.hh b/lix/libstore/remote-store.hh index 8c730cd4f..ec74060ae 100644 --- a/lix/libstore/remote-store.hh +++ b/lix/libstore/remote-store.hh @@ -97,7 +97,7 @@ public: const StorePathSet & references = StorePathSet() ) override; - kj::Promise> addToStore(const ValidPathInfo & info, Source & nar, + kj::Promise> addToStore(const ValidPathInfo & info, AsyncInputStream & nar, RepairFlag repair, CheckSigsFlag checkSigs) override; kj::Promise> addMultipleToStore( diff --git a/lix/libstore/store-api.cc b/lix/libstore/store-api.cc index ed06c382e..1209d545d 100644 --- a/lix/libstore/store-api.cc +++ b/lix/libstore/store-api.cc @@ -464,7 +464,7 @@ try { info.narSize = narSize; if (!isValidPath(info.path)) { - auto source = GeneratorSource{dumpPath(srcPath)}; + auto source = AsyncGeneratorInputStream{dumpPath(srcPath)}; TRY_AWAIT(addToStore(info, source)); } @@ -1054,25 +1054,27 @@ static std::string makeCopyPathMessage( namespace { -struct CopyPathSource : Source +struct CopyPathStream : AsyncInputStream { Activity & act; size_t copied = 0, expected; - box_ptr inner; + box_ptr inner; - CopyPathSource(Activity & act, size_t expected, box_ptr inner) + CopyPathStream(Activity & act, size_t expected, box_ptr inner) : act(act) , expected(expected) , inner(std::move(inner)) { } - size_t read(char * data, size_t len) override - { - auto result = inner->read(data, len); + kj::Promise> read(void * data, size_t len) override + try { + auto result = TRY_AWAIT(inner->read(data, len)); copied += result; act.progress(copied, expected); - return result; + co_return result; + } catch (...) { + co_return result::current_exception(); } }; } @@ -1116,7 +1118,11 @@ try { info = info2; } - CopyPathSource source{act, info->narSize, srcStore.narFromPath(storePath)}; + CopyPathStream source{ + act, + info->narSize, + make_box_ptr(srcStore.narFromPath(storePath)) + }; TRY_AWAIT(dstStore.addToStore(*info, source, repair, checkSigs)); co_return result::success(); } catch (...) { @@ -1234,22 +1240,25 @@ try { ValidPathInfo infoForDst = *info; infoForDst.path = storePathForDst; - struct SinglePathSource : CopyPathSource + struct SinglePathStream : CopyPathStream { Activity act; PushActivity pact{act.id}; - SinglePathSource( - std::string message, Logger::Fields fields, size_t expected, box_ptr inner + SinglePathStream( + std::string message, + Logger::Fields fields, + size_t expected, + box_ptr inner ) - : CopyPathSource(act, expected, std::move(inner)) + : CopyPathStream(act, expected, std::move(inner)) , act(*logger, lvlInfo, actCopyPath, message, fields) { } }; auto source = [](auto & srcStore, auto & dstStore, auto missingPath, auto info - ) -> kj::Promise>> { + ) -> kj::Promise>> { try { // We can reasonably assume that the copy will happen whenever we // read the path, so log something about that at that point @@ -1257,11 +1266,11 @@ try { auto dstUri = dstStore.getUri(); auto storePathS = srcStore.printStorePath(missingPath); - co_return make_box_ptr( + co_return make_box_ptr( makeCopyPathMessage(srcUri, dstUri, storePathS), Logger::Fields{storePathS, srcUri, dstUri}, info->narSize, - srcStore.narFromPath(missingPath) + make_box_ptr(srcStore.narFromPath(missingPath)) ); } catch (...) { co_return result::current_exception(); diff --git a/lix/libstore/store-api.hh b/lix/libstore/store-api.hh index 3ee46c4ed..a0171738d 100644 --- a/lix/libstore/store-api.hh +++ b/lix/libstore/store-api.hh @@ -500,15 +500,19 @@ public: /** * Import a path into the store. */ - virtual kj::Promise> addToStore(const ValidPathInfo & info, Source & narSource, - RepairFlag repair = NoRepair, CheckSigsFlag checkSigs = CheckSigs) = 0; + virtual kj::Promise> addToStore( + const ValidPathInfo & info, + AsyncInputStream & narSource, + RepairFlag repair = NoRepair, + CheckSigsFlag checkSigs = CheckSigs + ) = 0; /** * A list of paths infos along with a source providing the content * of the associated store path */ using PathsSource = std::vector< - std::pair>>()>>>; + std::pair>>()>>>; /** * Import multiple paths into the store. diff --git a/lix/nix/add-to-store.cc b/lix/nix/add-to-store.cc index ca4bfc04d..f58e7fcba 100644 --- a/lix/nix/add-to-store.cc +++ b/lix/nix/add-to-store.cc @@ -2,6 +2,7 @@ #include "lix/libmain/common-args.hh" #include "lix/libstore/store-api.hh" #include "lix/libutil/archive.hh" +#include "lix/libutil/async-io.hh" using namespace nix; @@ -54,7 +55,7 @@ struct CmdAddToStore : MixDryRun, StoreCommand info.narSize = sink.s.size(); if (!dryRun) { - auto source = StringSource(sink.s); + auto source = AsyncStringInputStream(sink.s); aio().blockOn(store->addToStore(info, source)); }