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

libstore: pass async streams to Store::addToStoreFromDump

Browse Source 
Change-Id: I1b1f7a63f093e33b0f8387e61342615708875ba7
This commit is contained in:
eldritch horrors
2025-03-03 20:48:59 +01:00
parent 6ba538a66f
commit dc8a634d24
11 changed files with 92 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/libfetchers/path.cc
+2 -1
View File
@@ -2,6 +2,7 @@
#include "lix/libfetchers/builtin-fetchers.hh"
#include "lix/libstore/store-api.hh"
#include "lix/libutil/archive.hh"
#include "lix/libutil/async-io.hh"
namespace nix::fetchers {
@@ -132,7 +133,7 @@ struct PathInputScheme : InputScheme
time_t mtime = 0;
if (!storePath || storePath->name() != "source" || !store->isValidPath(*storePath)) {
// FIXME: try to substitute storePath.
auto src = GeneratorSource{dumpPathAndGetMtime(absPath, mtime)};
auto src = AsyncGeneratorInputStream{dumpPathAndGetMtime(absPath, mtime)};
storePath = TRY_AWAIT(store->addToStoreFromDump(src, "source"));
}
input.attrs.insert_or_assign("lastModified", uint64_t(mtime));
lix/libstore/binary-cache-store.cc
+9 -4
View File
@@ -289,13 +289,18 @@ try {
co_return result::current_exception();
}
kj::Promise<Result<StorePath>> BinaryCacheStore::addToStoreFromDump(Source & dump, std::string_view name,
FileIngestionMethod method, HashType hashAlgo, RepairFlag repair, const StorePathSet & references)
kj::Promise<Result<StorePath>> BinaryCacheStore::addToStoreFromDump(
AsyncInputStream & dump,
std::string_view name,
FileIngestionMethod method,
HashType hashAlgo,
RepairFlag repair,
const StorePathSet & references
)
try {
if (method != FileIngestionMethod::Recursive || hashAlgo != HashType::SHA256)
unsupported("addToStoreFromDump");
AsyncSourceInputStream stream{dump};
co_return TRY_AWAIT(addToStoreCommon(stream, repair, CheckSigs, [&](HashResult nar) {
co_return TRY_AWAIT(addToStoreCommon(dump, repair, CheckSigs, [&](HashResult nar) {
ValidPathInfo info {
*this,
name,
lix/libstore/binary-cache-store.hh
+8 -2
View File
@@ -118,8 +118,14 @@ public:
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & narSource,
RepairFlag repair, CheckSigsFlag checkSigs) override;
kj::Promise<Result<StorePath>> addToStoreFromDump(Source & dump, std::string_view name,
FileIngestionMethod method, HashType hashAlgo, RepairFlag repair, const StorePathSet & references) override;
kj::Promise<Result<StorePath>> addToStoreFromDump(
AsyncInputStream & dump,
std::string_view name,
FileIngestionMethod method,
HashType hashAlgo,
RepairFlag repair,
const StorePathSet & references
) override;
kj::Promise<Result<StorePath>> addToStoreRecursive(
std::string_view name,
lix/libstore/build/local-derivation-goal.cc
+2 -1
View File
@@ -1,4 +1,5 @@
#include "lix/libstore/build/local-derivation-goal.hh"
#include "lix/libutil/async-io.hh"
#include "lix/libutil/async.hh"
#include "lix/libutil/error.hh"
#include "lix/libstore/indirect-root-store.hh"
@@ -1119,7 +1120,7 @@ struct RestrictedStore : public virtual IndirectRootStore, public virtual GcStor
}
kj::Promise<Result<StorePath>> addToStoreFromDump(
Source & dump,
AsyncInputStream & dump,
std::string_view name,
FileIngestionMethod method,
HashType hashAlgo,
lix/libstore/daemon.cc
+6 -2
View File
@@ -1,4 +1,5 @@
#include "lix/libstore/daemon.hh"
#include "lix/libutil/async-io.hh"
#include "lix/libutil/monitor-fd.hh"
#include "lix/libstore/worker-protocol.hh"
#include "lix/libstore/worker-protocol-impl.hh"
@@ -428,7 +429,10 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
return store->queryPathInfo(path);
},
[&](const FileIngestionMethod & fim) {
auto path = aio.blockOn(store->addToStoreFromDump(source, name, fim, hashType, repair, refs));
AsyncSourceInputStream stream{source};
auto path = aio.blockOn(
store->addToStoreFromDump(stream, name, fim, hashType, repair, refs)
);
return store->queryPathInfo(path);
},
}, contentAddressMethod.raw);
@@ -495,7 +499,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
co_yield std::move(file->contents);
}
};
GeneratorSource dumpSource{g()};
AsyncGeneratorInputStream dumpSource{g()};
logger->startWork();
auto path = aio.blockOn(store->addToStoreFromDump(dumpSource, baseName, method, hashAlgo));
logger->stopWork();
lix/libstore/local-store.cc
+27 -19
View File
@@ -6,6 +6,7 @@
#include "lix/libstore/worker-protocol.hh"
#include "lix/libstore/derivations.hh"
#include "lix/libstore/nar-info.hh"
#include "lix/libutil/async-io.hh"
#include "lix/libutil/async.hh"
#include "lix/libutil/references.hh"
#include "lix/libutil/result.hh"
@@ -1287,12 +1288,18 @@ try {
}
kj::Promise<Result<StorePath>> LocalStore::addToStoreFromDump(Source & source0, std::string_view name,
FileIngestionMethod method, HashType hashAlgo, RepairFlag repair, const StorePathSet & references)
kj::Promise<Result<StorePath>> LocalStore::addToStoreFromDump(
AsyncInputStream & source0,
std::string_view name,
FileIngestionMethod method,
HashType hashAlgo,
RepairFlag repair,
const StorePathSet & references
)
try {
/* For computing the store path. */
auto hashSink = std::make_unique<HashSink>(hashAlgo);
TeeSource source { source0, *hashSink };
AsyncTeeInputStream source { source0, *hashSink };
/* Read the source path into memory, but only if it's up to
narBufferSize bytes. If it's larger, write it to a temporary
@@ -1327,9 +1334,8 @@ try {
Finally cleanup([&]() {
dump = {dumpBuffer.get(), dump.size() + got};
});
try {
got = source.read(dumpBuffer.get() + oldSize, want);
} catch (EndOfFile &) {
got = TRY_AWAIT(source.read(dumpBuffer.get() + oldSize, want));
if (got == 0) {
inMemory = true;
break;
}
@@ -1341,29 +1347,31 @@ try {
AutoCloseFD tempDirFd;
if (!inMemory) {
struct ChainSource : Source
struct ChainSource : AsyncInputStream
{
Source & source1, & source2;
AsyncInputStream & source1, & source2;
bool useSecond = false;
ChainSource(Source & s1, Source & s2) : source1(s1), source2(s2) {}
ChainSource(AsyncInputStream & s1, AsyncInputStream & s2) : source1(s1), source2(s2) {}
size_t read(char * data, size_t len) override
{
kj::Promise<Result<size_t>> read(void * data, size_t len) override
try {
if (useSecond) {
return source2.read(data, len);
co_return TRY_AWAIT(source2.read(data, len));
} else {
try {
return source1.read(data, len);
} catch (EndOfFile &) {
if (auto got = TRY_AWAIT(source1.read(data, len))) {
co_return got;
} else {
useSecond = true;
return this->read(data, len);
co_return TRY_AWAIT(read(data, len));
}
}
} catch (...) {
co_return result::current_exception();
}
};
/* Drain what we pulled so far, and then keep on pulling */
StringSource dumpSource { dump };
AsyncStringInputStream dumpSource { dump };
ChainSource bothSource { dumpSource, source };
std::tie(tempDir, tempDirFd) = createTempDirInStore();
@@ -1371,9 +1379,9 @@ try {
tempPath = tempDir + "/x";
if (method == FileIngestionMethod::Recursive)
restorePath(tempPath, bothSource);
TRY_AWAIT(restorePath(tempPath, bothSource));
else
writeFile(tempPath, bothSource);
TRY_AWAIT(writeFile(tempPath, bothSource));
dumpBuffer.reset();
dump = {};
lix/libstore/local-store.hh
+9 -2
View File
@@ -5,6 +5,7 @@
#include "lix/libstore/store-api.hh"
#include "lix/libstore/indirect-root-store.hh"
#include "lix/libutil/async-io.hh"
#include "lix/libutil/sync.hh"
#include "lix/libutil/types.hh"
@@ -210,8 +211,14 @@ public:
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & source,
RepairFlag repair, CheckSigsFlag checkSigs) override;
kj::Promise<Result<StorePath>> addToStoreFromDump(Source & dump, std::string_view name,
FileIngestionMethod method, HashType hashAlgo, RepairFlag repair, const StorePathSet & references) override;
kj::Promise<Result<StorePath>> addToStoreFromDump(
AsyncInputStream & dump,
std::string_view name,
FileIngestionMethod method,
HashType hashAlgo,
RepairFlag repair,
const StorePathSet & references
) override;
kj::Promise<Result<StorePath>> addTextToStore(
std::string_view name,
lix/libstore/remote-store.cc
+9 -4
View File
@@ -488,11 +488,16 @@ try {
}
kj::Promise<Result<StorePath>> RemoteStore::addToStoreFromDump(Source & dump, std::string_view name,
FileIngestionMethod method, HashType hashType, RepairFlag repair, const StorePathSet & references)
kj::Promise<Result<StorePath>> RemoteStore::addToStoreFromDump(
AsyncInputStream & dump,
std::string_view name,
FileIngestionMethod method,
HashType hashType,
RepairFlag repair,
const StorePathSet & references
)
try {
AsyncSourceInputStream stream{dump};
co_return TRY_AWAIT(addCAToStore(stream, name, method, hashType, references, repair))->path;
co_return TRY_AWAIT(addCAToStore(dump, name, method, hashType, references, repair))->path;
} catch (...) {
co_return result::current_exception();
}
lix/libstore/remote-store.hh
+8 -2
View File
@@ -88,8 +88,14 @@ public:
/**
* Add a content-addressable store path. Does not support references. `dump` will be drained.
*/
kj::Promise<Result<StorePath>> addToStoreFromDump(Source & dump, std::string_view name,
FileIngestionMethod method = FileIngestionMethod::Recursive, HashType hashAlgo = HashType::SHA256, RepairFlag repair = NoRepair, const StorePathSet & references = StorePathSet()) override;
kj::Promise<Result<StorePath>> addToStoreFromDump(
AsyncInputStream & dump,
std::string_view name,
FileIngestionMethod method = FileIngestionMethod::Recursive,
HashType hashAlgo = HashType::SHA256,
RepairFlag repair = NoRepair,
const StorePathSet & references = StorePathSet()
) override;
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & nar,
RepairFlag repair, CheckSigsFlag checkSigs) override;
lix/libstore/store-api.cc
+3 -2
View File
@@ -3,6 +3,7 @@
#include "lix/libstore/derivations.hh"
#include "lix/libstore/store-api.hh"
#include "lix/libstore/nar-info-disk-cache.hh"
#include "lix/libutil/async-io.hh"
#include "lix/libutil/async.hh"
#include "lix/libutil/box_ptr.hh"
#include "lix/libutil/hash.hh"
@@ -286,7 +287,7 @@ kj::Promise<Result<StorePath>> Store::addToStoreRecursive(
HashType hashAlgo,
RepairFlag repair)
try {
auto source = GeneratorSource{_source.dump()};
auto source = AsyncGeneratorInputStream{_source.dump()};
co_return TRY_AWAIT(
addToStoreFromDump(source, name, FileIngestionMethod::Recursive, hashAlgo, repair, {})
);
@@ -301,7 +302,7 @@ kj::Promise<Result<StorePath>> Store::addToStoreFlat(
RepairFlag repair)
try {
Path srcPath(absPath(_srcPath));
auto source = GeneratorSource{readFileSource(srcPath)};
auto source = AsyncGeneratorInputStream{readFileSource(srcPath)};
co_return TRY_AWAIT(
addToStoreFromDump(source, name, FileIngestionMethod::Flat, hashAlgo, repair, {})
);
lix/libstore/store-api.hh
+9 -3
View File
@@ -2,6 +2,7 @@
///@file
#include "lix/libutil/archive.hh"
#include "lix/libutil/async-io.hh"
#include "lix/libutil/async.hh"
#include "lix/libutil/logging.hh"
#include "lix/libstore/nar-info.hh"
@@ -555,9 +556,14 @@ public:
*
* \todo remove?
*/
virtual kj::Promise<Result<StorePath>> addToStoreFromDump(Source & dump, std::string_view name,
FileIngestionMethod method = FileIngestionMethod::Recursive, HashType hashAlgo = HashType::SHA256, RepairFlag repair = NoRepair,
const StorePathSet & references = StorePathSet())
virtual kj::Promise<Result<StorePath>> addToStoreFromDump(
AsyncInputStream & dump,
std::string_view name,
FileIngestionMethod method = FileIngestionMethod::Recursive,
HashType hashAlgo = HashType::SHA256,
RepairFlag repair = NoRepair,
const StorePathSet & references = StorePathSet()
)
try { unsupported("addToStoreFromDump"); } catch (...) { return {result::current_exception()}; }
/**