guiwf/lix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

libstore: pass async streams to Store::add{,Multiple}ToStore

Browse Source 
Change-Id: Idafc0d640bf66d2d28ff71ae81546db8cf463ee2
This commit is contained in:
eldritch horrors
2025-03-03 21:13:44 +01:00
parent dc8a634d24
commit 14067da947
18 changed files with 98 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/legacy/nix-store.cc
+3 -1
View File
@@ -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();
lix/libcmd/cmd-profiles.cc
+2 -1
View File
@@ -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);
lix/libfetchers/tarball.cc
+2 -1
View File
@@ -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);
}
lix/libstore/binary-cache-store.cc
+8 -5
View File
@@ -267,17 +267,20 @@ try {
co_return result::current_exception();
}
kj::Promise<Result<void>> BinaryCacheStore::addToStore(const ValidPathInfo & info, Source & narSource,
RepairFlag repair, CheckSigsFlag checkSigs)
kj::Promise<Result<void>> 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);
lix/libstore/binary-cache-store.hh
+1 -1
View File
@@ -115,7 +115,7 @@ public:
std::optional<StorePath> queryPathFromHashPart(const std::string & hashPart) override;
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & narSource,
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, AsyncInputStream & narSource,
RepairFlag repair, CheckSigsFlag checkSigs) override;
kj::Promise<Result<StorePath>> addToStoreFromDump(
lix/libstore/build/local-derivation-goal.cc
+1 -1
View File
@@ -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<Result<void>> addToStore(const ValidPathInfo & info, Source & narSource,
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, AsyncInputStream & narSource,
RepairFlag repair = NoRepair, CheckSigsFlag checkSigs = CheckSigs) override
try {
TRY_AWAIT(next->addToStore(info, narSource, repair, checkSigs));
lix/libstore/daemon.cc
+6 -3
View File
@@ -524,8 +524,9 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> 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> 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> 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();
lix/libstore/dummy-store.cc
+1 -1
View File
@@ -54,7 +54,7 @@ struct DummyStore final : public Store
std::optional<StorePath> queryPathFromHashPart(const std::string & hashPart) override
{ unsupported("queryPathFromHashPart"); }
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & source,
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, AsyncInputStream & source,
RepairFlag repair, CheckSigsFlag checkSigs) override
try { unsupported("addToStore"); } catch (...) { return {result::current_exception()}; }
lix/libstore/export-import.cc
+1 -1
View File
@@ -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);
lix/libstore/legacy-ssh-store.cc
+3 -3
View File
@@ -189,7 +189,7 @@ struct LegacySSHStore final : public Store
return info;
}
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & source,
kj::Promise<Result<void>> 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;
lix/libstore/local-store.cc
+11 -6
View File
@@ -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<Result<void>> LocalStore::addToStore(const ValidPathInfo & info, Source & source,
RepairFlag repair, CheckSigsFlag checkSigs)
kj::Promise<Result<void>> 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 (...) {
lix/libstore/local-store.hh
+1 -1
View File
@@ -208,7 +208,7 @@ public:
bool pathInfoIsUntrusted(const ValidPathInfo &) override;
bool realisationIsUntrusted(const Realisation & ) override;
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & source,
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, AsyncInputStream & source,
RepairFlag repair, CheckSigsFlag checkSigs) override;
kj::Promise<Result<StorePath>> addToStoreFromDump(
lix/libstore/make-content-addressed.cc
+2 -1
View File
@@ -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));
lix/libstore/remote-store.cc
+21 -7
View File
@@ -503,8 +503,12 @@ try {
}
kj::Promise<Result<void>> RemoteStore::addToStore(const ValidPathInfo & info, Source & source,
RepairFlag repair, CheckSigsFlag checkSigs)
kj::Promise<Result<void>> 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<void>();
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<ValidPathInfo>::write(*this,
WorkerProto::WriteConn {remoteVersion},
pathInfo);
TRY_AWAIT(pathSource())->drainInto(sink);
TRY_AWAIT(TRY_AWAIT(pathSource())->drainInto(sink));
}
co_return result::success();
} catch (...) {
lix/libstore/remote-store.hh
+1 -1
View File
@@ -97,7 +97,7 @@ public:
const StorePathSet & references = StorePathSet()
) override;
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & nar,
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, AsyncInputStream & nar,
RepairFlag repair, CheckSigsFlag checkSigs) override;
kj::Promise<Result<void>> addMultipleToStore(
lix/libstore/store-api.cc
+25 -16
View File
@@ -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<Source> inner;
box_ptr<AsyncInputStream> inner;
CopyPathSource(Activity & act, size_t expected, box_ptr<Source> inner)
CopyPathStream(Activity & act, size_t expected, box_ptr<AsyncInputStream> 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<Result<size_t>> 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<AsyncSourceInputStream>(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<Source> inner
SinglePathStream(
std::string message,
Logger::Fields fields,
size_t expected,
box_ptr<AsyncInputStream> 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<Result<box_ptr<Source>>> {
) -> kj::Promise<Result<box_ptr<AsyncInputStream>>> {
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<SinglePathSource>(
co_return make_box_ptr<SinglePathStream>(
makeCopyPathMessage(srcUri, dstUri, storePathS),
Logger::Fields{storePathS, srcUri, dstUri},
info->narSize,
srcStore.narFromPath(missingPath)
make_box_ptr<AsyncSourceInputStream>(srcStore.narFromPath(missingPath))
);
} catch (...) {
co_return result::current_exception();
lix/libstore/store-api.hh
+7 -3
View File
@@ -500,15 +500,19 @@ public:
/**
* Import a path into the store.
*/
virtual kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & narSource,
RepairFlag repair = NoRepair, CheckSigsFlag checkSigs = CheckSigs) = 0;
virtual kj::Promise<Result<void>> 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<ValidPathInfo, std::function<kj::Promise<Result<box_ptr<Source>>>()>>>;
std::pair<ValidPathInfo, std::function<kj::Promise<Result<box_ptr<AsyncInputStream>>>()>>>;
/**
* Import multiple paths into the store.
lix/nix/add-to-store.cc
+2 -1
View File
@@ -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));
}