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

libstore: asyncify Store::narFromPath return stream

Browse Source 
Change-Id: I051c58e650109c70021c0e0a745c7342226e295b
This commit is contained in:
eldritch horrors
2025-06-15 13:36:31 +00:00
parent d824753377
commit 743703ce35
17 changed files with 66 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/legacy/nix-store.cc
+3 -3
View File
@@ -800,7 +800,7 @@ static void opVerifyPath(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
printMsg(lvlTalkative, "checking path '%s'...", store->printStorePath(path));
auto info = aio.blockOn(store->queryPathInfo(path));
HashSink sink(info->narHash.type);
aio.blockOn(store->narFromPath(path))->drainInto(sink);
aio.blockOn(aio.blockOn(store->narFromPath(path))->drainInto(sink));
auto current = sink.finish();
if (current.first != info->narHash) {
printError("path '%s' was modified! expected hash '%s', got '%s'",
@@ -939,8 +939,8 @@ static void opServe(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
}
case ServeProto::Command::DumpStorePath:
aio.blockOn(store->narFromPath(store->parseStorePath(readString(in))))
->drainInto(out);
aio.blockOn(aio.blockOn(store->narFromPath(store->parseStorePath(readString(in))))
->drainInto(out));
break;
case ServeProto::Command::ImportPaths: {
lix/libstore/binary-cache-store.cc
+38 -21
View File
@@ -1,6 +1,8 @@
#include "lix/libutil/archive.hh"
#include "lix/libstore/binary-cache-store.hh"
#include "lix/libutil/async-io.hh"
#include "lix/libutil/async.hh"
#include "lix/libutil/box_ptr.hh"
#include "lix/libutil/compression.hh"
#include "lix/libstore/derivations.hh"
#include "lix/libstore/fs-accessor.hh"
@@ -362,34 +364,49 @@ try {
co_return result::current_exception();
}
kj::Promise<Result<box_ptr<Source>>> BinaryCacheStore::narFromPath(const StorePath & storePath)
kj::Promise<Result<box_ptr<AsyncInputStream>>>
BinaryCacheStore::narFromPath(const StorePath & storePath)
try {
struct NarFromPath : AsyncInputStream
{
Stats<std::atomic> & stats;
box_ptr<Source> file;
box_ptr<AsyncInputStream> decompressed;
uint64_t total;
NarFromPath(Stats<std::atomic> & stats, const std::string & method, box_ptr<Source> file)
: stats(stats)
, file(std::move(file))
, decompressed(
makeDecompressionStream(method, make_box_ptr<AsyncSourceInputStream>(*this->file))
)
{
}
kj::Promise<Result<size_t>> read(void * buffer, size_t size) override
{
return decompressed->read(buffer, size).then([&](auto r) {
if (r.has_value()) {
if (r.value() > 0) {
total += r.value();
} else {
stats.narRead++;
// stats.narReadCompressedBytes += nar->size(); // FIXME
stats.narReadBytes += total;
}
}
return r;
});
}
};
auto info_ = TRY_AWAIT(queryPathInfo(storePath)).try_cast<const NarInfo>();
assert(info_ && "binary cache queryPathInfo didn't return a NarInfo");
auto & info = *info_;
try {
auto file = getFile(info->url);
co_return make_box_ptr<GeneratorSource>(
[](auto info, auto file, auto & stats) -> WireFormatGenerator {
constexpr size_t buflen = 65536;
auto buf = std::make_unique<char[]>(buflen);
size_t total = 0;
auto decompressor = makeDecompressionSource(info->compression, *file);
try {
while (true) {
const auto len = decompressor->read(buf.get(), buflen);
co_yield std::span{buf.get(), len};
total += len;
}
} catch (EndOfFile &) {
}
stats.narRead++;
// stats.narReadCompressedBytes += nar->size(); // FIXME
stats.narReadBytes += total;
}(std::move(info), std::move(file), stats)
);
co_return make_box_ptr<NarFromPath>(stats, info->compression, std::move(file));
} catch (NoSuchBinaryCacheFile & e) {
throw SubstituteGone(std::move(e.info()));
}
lix/libstore/binary-cache-store.hh
+1 -1
View File
@@ -146,7 +146,7 @@ public:
const StorePathSet & references,
RepairFlag repair) override;
kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override;
kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override;
ref<FSAccessor> getFSAccessor() override;
lix/libstore/dummy-store.cc
+2 -1
View File
@@ -1,5 +1,6 @@
#include "lix/libstore/dummy-store.hh"
#include "lix/libstore/store-api.hh"
#include "lix/libutil/async-io.hh"
namespace nix {
@@ -71,7 +72,7 @@ struct DummyStore final : public Store
RepairFlag repair) override
try { unsupported("addTextToStore"); } catch (...) { return {result::current_exception()}; }
kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override
kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override
try { unsupported("narFromPath"); } catch (...) { return {result::current_exception()}; }
virtual ref<FSAccessor> getFSAccessor() override
lix/libstore/export-import.cc
+1 -1
View File
@@ -32,7 +32,7 @@ try {
HashSink hashSink(HashType::SHA256);
TeeSink teeSink(sink, hashSink);
TRY_AWAIT(narFromPath(path))->drainInto(teeSink);
TRY_AWAIT(TRY_AWAIT(narFromPath(path))->drainInto(teeSink));
/* Refuse to export paths that have changed. This prevents
filesystem corruption from spreading to other machines.
lix/libstore/legacy-ssh-store.cc
+3 -2
View File
@@ -1,5 +1,6 @@
#include "lix/libstore/legacy-ssh-store.hh"
#include "lix/libutil/archive.hh"
#include "lix/libutil/async-io.hh"
#include "lix/libutil/pool.hh"
#include "lix/libstore/remote-store.hh"
#include "lix/libstore/serve-protocol.hh"
@@ -255,13 +256,13 @@ struct LegacySSHStore final : public Store
co_return result::current_exception();
}
kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override
kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override
try {
auto conn(TRY_AWAIT(connections->get()));
conn->to << ServeProto::Command::DumpStorePath << printStorePath(path);
conn->to.flush();
co_return make_box_ptr<GeneratorSource>([] (auto conn) -> WireFormatGenerator {
co_return make_box_ptr<AsyncGeneratorInputStream>([](auto conn) -> WireFormatGenerator {
co_yield copyNAR(conn->from);
}(std::move(conn)));
} catch (...) {
lix/libstore/local-fs-store.cc
+3 -2
View File
@@ -3,6 +3,7 @@
#include "lix/libstore/store-api.hh"
#include "lix/libstore/local-fs-store.hh"
#include "lix/libstore/globals.hh"
#include "lix/libutil/async-io.hh"
#include "lix/libutil/compression.hh"
#include "lix/libstore/derivations.hh"
@@ -84,11 +85,11 @@ ref<FSAccessor> LocalFSStore::getFSAccessor()
std::dynamic_pointer_cast<LocalFSStore>(shared_from_this())));
}
kj::Promise<Result<box_ptr<Source>>> LocalFSStore::narFromPath(const StorePath & path)
kj::Promise<Result<box_ptr<AsyncInputStream>>> LocalFSStore::narFromPath(const StorePath & path)
try {
if (!TRY_AWAIT(isValidPath(path)))
throw Error("path '%s' does not exist in store", printStorePath(path));
co_return make_box_ptr<GeneratorSource>(
co_return make_box_ptr<AsyncGeneratorInputStream>(
dumpPath(getRealStoreDir() + std::string(printStorePath(path), config().storeDir.size()))
);
} catch (...) {
lix/libstore/local-fs-store.hh
+2 -1
View File
@@ -4,6 +4,7 @@
#include "lix/libstore/store-api.hh"
#include "lix/libstore/gc-store.hh"
#include "lix/libstore/log-store.hh"
#include "lix/libutil/async-io.hh"
namespace nix {
@@ -42,7 +43,7 @@ public:
LocalFSStoreConfig & config() override = 0;
const LocalFSStoreConfig & config() const override = 0;
kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override;
kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override;
ref<FSAccessor> getFSAccessor() override;
/**
lix/libstore/make-content-addressed.cc
+1 -1
View File
@@ -26,7 +26,7 @@ try {
std::string oldHashPart(path.hashPart());
StringSink sink;
TRY_AWAIT(srcStore.narFromPath(path))->drainInto(sink);
TRY_AWAIT(TRY_AWAIT(srcStore.narFromPath(path))->drainInto(sink));
StringMap rewrites;
lix/libstore/remote-fs-accessor.cc
+1 -1
View File
@@ -101,7 +101,7 @@ try {
}
StringSink sink;
TRY_AWAIT(store->narFromPath(storePath))->drainInto(sink);
TRY_AWAIT(TRY_AWAIT(store->narFromPath(storePath))->drainInto(sink));
co_return {TRY_AWAIT(addToCache(storePath.hashPart(), std::move(sink.s))), restPath};
} catch (...) {
co_return result::current_exception();
lix/libstore/remote-store.cc
+2 -2
View File
@@ -719,11 +719,11 @@ RemoteStore::Connection::~Connection()
}
}
kj::Promise<Result<box_ptr<Source>>> RemoteStore::narFromPath(const StorePath & path)
kj::Promise<Result<box_ptr<AsyncInputStream>>> RemoteStore::narFromPath(const StorePath & path)
try {
auto conn(TRY_AWAIT(getConnection()));
TRY_AWAIT(conn.sendCommand(WorkerProto::Op::NarFromPath, printStorePath(path)));
co_return make_box_ptr<GeneratorSource>([](auto conn) -> WireFormatGenerator {
co_return make_box_ptr<AsyncGeneratorInputStream>([](auto conn) -> WireFormatGenerator {
co_yield copyNAR(conn->from);
}(std::move(conn)));
} catch (...) {
lix/libstore/remote-store.hh
+2 -1
View File
@@ -197,7 +197,8 @@ protected:
virtual ref<FSAccessor> getFSAccessor() override;
virtual kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override;
virtual kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path
) override;
private:
lix/libstore/store-api.cc
+2 -6
View File
@@ -1071,11 +1071,7 @@ try {
info = info2;
}
CopyPathStream source{
act,
info->narSize,
make_box_ptr<AsyncSourceInputStream>(TRY_AWAIT(srcStore.narFromPath(storePath)))
};
CopyPathStream source{act, info->narSize, TRY_AWAIT(srcStore.narFromPath(storePath))};
TRY_AWAIT(dstStore.addToStore(*info, source, repair, checkSigs));
co_return result::success();
} catch (...) {
@@ -1187,7 +1183,7 @@ try {
makeCopyPathMessage(srcUri, dstUri, storePathS),
Logger::Fields{storePathS, srcUri, dstUri},
info->narSize,
make_box_ptr<AsyncSourceInputStream>(TRY_AWAIT(srcStore.narFromPath(missingPath)))
TRY_AWAIT(srcStore.narFromPath(missingPath))
);
} catch (...) {
co_return result::current_exception();
lix/libstore/store-api.hh
+1 -1
View File
@@ -575,7 +575,7 @@ public:
/**
* Generate a NAR dump of a store path.
*/
virtual kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) = 0;
virtual kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) = 0;
/**
* For each path, if it's a derivation, build it. Building a
lix/libstore/uds-remote-store.hh
+2 -1
View File
@@ -4,6 +4,7 @@
#include "lix/libstore/remote-store.hh"
#include "lix/libstore/remote-store-connection.hh"
#include "lix/libstore/indirect-root-store.hh"
#include "lix/libutil/async-io.hh"
namespace nix {
@@ -42,7 +43,7 @@ public:
ref<FSAccessor> getFSAccessor() override
{ return LocalFSStore::getFSAccessor(); }
kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override
kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override
{ return LocalFSStore::narFromPath(path); }
/**
lix/nix/dump-path.cc
+1 -1
View File
@@ -23,7 +23,7 @@ struct CmdDumpPath : StorePathCommand
{
logger->pause();
FdSink sink(STDOUT_FILENO);
aio().blockOn(store->narFromPath(storePath))->drainInto(sink);
aio().blockOn(aio().blockOn(store->narFromPath(storePath))->drainInto(sink));
sink.flush();
}
};
lix/nix/verify.cc
+1 -1
View File
@@ -102,7 +102,7 @@ struct CmdVerify : StorePathsCommand
auto hashSink = HashSink(info->narHash.type);
aio.blockOn(store->narFromPath(info->path))->drainInto(hashSink);
aio.blockOn(aio.blockOn(store->narFromPath(info->path))->drainInto(hashSink));
auto hash = hashSink.finish();