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

libstore: asyncify Store::addToStore

Browse Source 
Change-Id: I0da60f3dc74525395bbf75fe6db56fce572cca7c
This commit is contained in:
eldritch horrors
2025-02-10 12:54:19 +00:00
parent 6b86f0bf49
commit af43fdec70
23 changed files with 108 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/legacy/nix-store.cc
+7 -3
View File
@@ -177,8 +177,12 @@ static void opAdd(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
{
if (!opFlags.empty()) throw UsageError("unknown flag");
for (auto & i : opArgs)
cout << fmt("%s\n", store->printStorePath(store->addToStore(std::string(baseNameOf(i)), i)));
for (auto & i : opArgs) {
cout << fmt(
"%s\n",
store->printStorePath(aio.blockOn(store->addToStore(std::string(baseNameOf(i)), i)))
);
}
}
@@ -995,7 +999,7 @@ static void opServe(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
SizedSource sizedSource(in, info.narSize);
store->addToStore(info, sizedSource, NoRepair, NoCheckSigs);
aio.blockOn(store->addToStore(info, sizedSource, NoRepair, NoCheckSigs));
// consume all the data that has been sent before continuing.
sizedSource.drainAll();
lix/libcmd/cmd-profiles.cc
+1 -1
View File
@@ -267,7 +267,7 @@ try {
info.narSize = sink.s.size();
StringSource source(sink.s);
store->addToStore(info, source);
TRY_AWAIT(store->addToStore(info, source));
co_return std::move(info.path);
} catch (...) {
lix/libfetchers/fetch-to-store.cc
+1 -1
View File
@@ -19,7 +19,7 @@ try {
co_return
settings.readOnlyMode
? store.computeStorePathForPath(name, path.canonical().abs(), method, HashType::SHA256, filter2).first
: store.addToStore(name, path.canonical().abs(), method, HashType::SHA256, filter2, repair);
: TRY_AWAIT(store.addToStore(name, path.canonical().abs(), method, HashType::SHA256, filter2, repair));
} catch (...) {
co_return result::current_exception();
}
lix/libfetchers/git.cc
+12 -6
View File
@@ -207,8 +207,8 @@ WorkdirInfo getWorkdirInfo(const Input & input, const Path & workdir)
return WorkdirInfo { .clean = clean, .hasHead = hasHead };
}
std::pair<StorePath, Input> fetchFromWorkdir(ref<Store> store, Input & input, const Path & workdir, const WorkdirInfo & workdirInfo)
{
static kj::Promise<Result<std::pair<StorePath, Input>>> fetchFromWorkdir(ref<Store> store, Input & input, const Path & workdir, const WorkdirInfo & workdirInfo)
try {
const bool submodules = maybeGetBoolAttr(input.attrs, "submodules").value_or(false);
auto gitDir = ".git";
@@ -242,7 +242,9 @@ std::pair<StorePath, Input> fetchFromWorkdir(ref<Store> store, Input & input, co
return files.count(file);
};
auto storePath = store->addToStore(input.getName(), actualPath, FileIngestionMethod::Recursive, HashType::SHA256, filter);
auto storePath = TRY_AWAIT(store->addToStore(
input.getName(), actualPath, FileIngestionMethod::Recursive, HashType::SHA256, filter
));
// FIXME: maybe we should use the timestamp of the last
// modified dirty file?
@@ -257,7 +259,9 @@ std::pair<StorePath, Input> fetchFromWorkdir(ref<Store> store, Input & input, co
runProgram("git", true, { "-C", actualPath, "--git-dir", gitDir, "rev-parse", "--verify", "--short", "HEAD" })) + "-dirty");
}
return {std::move(storePath), input};
co_return {std::move(storePath), input};
} catch (...) {
co_return result::current_exception();
}
} // end namespace
@@ -503,7 +507,7 @@ struct GitInputScheme : InputScheme
if (!input.getRef() && !input.getRev() && isLocal) {
auto workdirInfo = getWorkdirInfo(input, actualUrl);
if (!workdirInfo.clean) {
co_return fetchFromWorkdir(store, input, actualUrl, workdirInfo);
co_return TRY_AWAIT(fetchFromWorkdir(store, input, actualUrl, workdirInfo));
}
}
@@ -766,7 +770,9 @@ struct GitInputScheme : InputScheme
unpackTarfile(*proc.getStdout(), tmpDir);
}
auto storePath = store->addToStore(name, tmpDir, FileIngestionMethod::Recursive, HashType::SHA256, filter);
auto storePath = TRY_AWAIT(store->addToStore(
name, tmpDir, FileIngestionMethod::Recursive, HashType::SHA256, filter
));
auto lastModified = std::stoull(runProgram("git", true, { "-C", repoDir, "--git-dir", gitDir, "log", "-1", "--format=%ct", "--no-show-signature", input.getRev()->gitRev() }));
lix/libfetchers/mercurial.cc
+8 -2
View File
@@ -202,7 +202,13 @@ struct MercurialInputScheme : InputScheme
return files.count(file);
};
auto storePath = store->addToStore(input.getName(), actualPath, FileIngestionMethod::Recursive, HashType::SHA256, filter);
auto storePath = TRY_AWAIT(store->addToStore(
input.getName(),
actualPath,
FileIngestionMethod::Recursive,
HashType::SHA256,
filter
));
co_return {std::move(storePath), input};
}
@@ -313,7 +319,7 @@ struct MercurialInputScheme : InputScheme
deletePath(tmpDir + "/.hg_archival.txt");
auto storePath = store->addToStore(name, tmpDir);
auto storePath = TRY_AWAIT(store->addToStore(name, tmpDir));
Attrs infoAttrs({
{"rev", input.getRev()->gitRev()},
lix/libfetchers/tarball.cc
+9 -2
View File
@@ -89,7 +89,7 @@ try {
};
info.narSize = sink.s.size();
auto source = StringSource { sink.s };
store->addToStore(info, source, NoRepair, NoCheckSigs);
TRY_AWAIT(store->addToStore(info, source, NoRepair, NoCheckSigs));
storePath = std::move(info.path);
}
@@ -161,7 +161,14 @@ try {
throw nix::Error("tarball '%s' contains an unexpected number of top-level files", url);
auto topDir = tmpDir + "/" + members.begin()->name;
lastModified = lstat(topDir).st_mtime;
unpackedStorePath = store->addToStore(name, topDir, FileIngestionMethod::Recursive, HashType::SHA256, defaultPathFilter, NoRepair);
unpackedStorePath = TRY_AWAIT(store->addToStore(
name,
topDir,
FileIngestionMethod::Recursive,
HashType::SHA256,
defaultPathFilter,
NoRepair
));
}
Attrs infoAttrs({
lix/libstore/binary-cache-store.cc
+12 -6
View File
@@ -4,6 +4,7 @@
#include "lix/libstore/derivations.hh"
#include "lix/libstore/fs-accessor.hh"
#include "lix/libstore/nar-info.hh"
#include "lix/libutil/result.hh"
#include "lix/libutil/sync.hh"
#include "lix/libstore/remote-fs-accessor.hh"
#include "lix/libstore/nar-info-disk-cache.hh" // IWYU pragma: keep
@@ -255,13 +256,13 @@ ref<const ValidPathInfo> BinaryCacheStore::addToStoreCommon(
return narInfo;
}
void BinaryCacheStore::addToStore(const ValidPathInfo & info, Source & narSource,
kj::Promise<Result<void>> BinaryCacheStore::addToStore(const ValidPathInfo & info, Source & narSource,
RepairFlag repair, CheckSigsFlag checkSigs)
{
try {
if (!repair && isValidPath(info.path)) {
// FIXME: copyNAR -> null sink
narSource.drain();
return;
co_return result::success();
}
addToStoreCommon(narSource, repair, checkSigs, {[&](HashResult nar) {
@@ -271,6 +272,9 @@ void BinaryCacheStore::addToStore(const ValidPathInfo & info, Source & narSource
// assert(info.narSize == nar.second);
return info;
}});
co_return result::success();
} catch (...) {
co_return result::current_exception();
}
StorePath BinaryCacheStore::addToStoreFromDump(Source & dump, std::string_view name,
@@ -365,7 +369,7 @@ std::shared_ptr<const ValidPathInfo> BinaryCacheStore::queryPathInfoUncached(con
return std::make_shared<NarInfo>(*this, *data, narInfoFile);
}
StorePath BinaryCacheStore::addToStore(
kj::Promise<Result<StorePath>> BinaryCacheStore::addToStore(
std::string_view name,
const Path & srcPath,
FileIngestionMethod method,
@@ -373,7 +377,7 @@ StorePath BinaryCacheStore::addToStore(
PathFilter & filter,
RepairFlag repair,
const StorePathSet & references)
{
try {
/* FIXME: Make BinaryCacheStore::addToStoreCommon support
non-recursive+sha256 so we can just use the default
implementation of this method in terms of addToStoreFromDump. */
@@ -387,7 +391,7 @@ StorePath BinaryCacheStore::addToStore(
auto h = sink.finish().first;
auto source = GeneratorSource{dumpPath(srcPath, filter)};
return addToStoreCommon(source, repair, CheckSigs, [&](HashResult nar) {
co_return addToStoreCommon(source, repair, CheckSigs, [&](HashResult nar) {
ValidPathInfo info {
*this,
name,
@@ -405,6 +409,8 @@ StorePath BinaryCacheStore::addToStore(
info.narSize = nar.second;
return info;
})->path;
} catch (...) {
co_return result::current_exception();
}
StorePath BinaryCacheStore::addTextToStore(
lix/libstore/binary-cache-store.hh
+2 -2
View File
@@ -113,13 +113,13 @@ public:
std::optional<StorePath> queryPathFromHashPart(const std::string & hashPart) override;
void addToStore(const ValidPathInfo & info, Source & narSource,
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & narSource,
RepairFlag repair, CheckSigsFlag checkSigs) override;
StorePath addToStoreFromDump(Source & dump, std::string_view name,
FileIngestionMethod method, HashType hashAlgo, RepairFlag repair, const StorePathSet & references) override;
StorePath addToStore(
kj::Promise<Result<StorePath>> addToStore(
std::string_view name,
const Path & srcPath,
FileIngestionMethod method,
lix/libstore/build/local-derivation-goal.cc
+8 -5
View File
@@ -1049,7 +1049,7 @@ struct RestrictedStore : public virtual IndirectRootStore, public virtual GcStor
std::optional<StorePath> queryPathFromHashPart(const std::string & hashPart) override
{ throw Error("queryPathFromHashPart"); }
StorePath addToStore(
kj::Promise<Result<StorePath>> addToStore(
std::string_view name,
const Path & srcPath,
FileIngestionMethod method,
@@ -1057,13 +1057,16 @@ struct RestrictedStore : public virtual IndirectRootStore, public virtual GcStor
PathFilter & filter,
RepairFlag repair,
const StorePathSet & references) override
{ throw Error("addToStore"); }
try { throw Error("addToStore"); } catch (...) { return {result::current_exception()}; }
void addToStore(const ValidPathInfo & info, Source & narSource,
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & narSource,
RepairFlag repair = NoRepair, CheckSigsFlag checkSigs = CheckSigs) override
{
next->addToStore(info, narSource, repair, checkSigs);
try {
TRY_AWAIT(next->addToStore(info, narSource, repair, checkSigs));
goal.addDependency(info.path);
co_return result::success();
} catch (...) {
co_return result::current_exception();
}
StorePath addTextToStore(
lix/libstore/build/substitution-goal.cc
+2
View File
@@ -210,6 +210,7 @@ try {
outPipe = kj::mv(pipe.fulfiller);
thr = std::async(std::launch::async, [this]() {
AsyncIoRoot aio;
/* Wake up the worker loop when we're done. */
Finally updateStats([this]() { outPipe->fulfill(); });
@@ -221,6 +222,7 @@ try {
PushActivity pact(act.id);
copyStorePath(
aio,
*sub,
worker.store,
fetchPath,
lix/libstore/daemon.cc
+4 -4
View File
@@ -900,8 +900,8 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
logger->startWork();
{
FramedSource source(from);
store->addToStore(info, source, (RepairFlag) repair,
dontCheckSigs ? NoCheckSigs : CheckSigs);
aio.blockOn(store->addToStore(info, source, (RepairFlag) repair,
dontCheckSigs ? NoCheckSigs : CheckSigs));
}
logger->stopWork();
}
@@ -913,8 +913,8 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
logger->startWork();
// FIXME: race if addToStore doesn't read source?
store->addToStore(info, *source, (RepairFlag) repair,
dontCheckSigs ? NoCheckSigs : CheckSigs);
aio.blockOn(store->addToStore(info, *source, (RepairFlag) repair,
dontCheckSigs ? NoCheckSigs : CheckSigs));
logger->stopWork();
}
lix/libstore/dummy-store.cc
+2 -2
View File
@@ -54,9 +54,9 @@ struct DummyStore final : public Store
std::optional<StorePath> queryPathFromHashPart(const std::string & hashPart) override
{ unsupported("queryPathFromHashPart"); }
void addToStore(const ValidPathInfo & info, Source & source,
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & source,
RepairFlag repair, CheckSigsFlag checkSigs) override
{ unsupported("addToStore"); }
try { unsupported("addToStore"); } catch (...) { return {result::current_exception()}; }
StorePath addTextToStore(
std::string_view name,
lix/libstore/export-import.cc
+1 -1
View File
@@ -86,7 +86,7 @@ try {
// Can't use underlying source, which would have been exhausted
auto source = StringSource(saved.s);
addToStore(info, source, NoRepair, checkSigs);
TRY_AWAIT(addToStore(info, source, NoRepair, checkSigs));
res.push_back(info.path);
}
lix/libstore/legacy-ssh-store.cc
+8 -4
View File
@@ -9,6 +9,7 @@
#include "lix/libstore/path-with-outputs.hh"
#include "lix/libstore/ssh.hh"
#include "lix/libstore/ssh-store.hh"
#include "lix/libutil/result.hh"
#include "lix/libutil/strings.hh"
#include "lix/libstore/derivations.hh"
@@ -185,9 +186,9 @@ struct LegacySSHStore final : public Store
return info;
}
void addToStore(const ValidPathInfo & info, Source & source,
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & source,
RepairFlag repair, CheckSigsFlag checkSigs) override
{
try {
debug("adding path '%s' to remote host '%s'", printStorePath(info.path), host);
auto conn(connections->get());
@@ -239,6 +240,9 @@ struct LegacySSHStore final : public Store
if (readInt(conn->from) != 1)
throw Error("failed to add path '%s' to remote host '%s'", printStorePath(info.path), host);
co_return result::success();
} catch (...) {
co_return result::current_exception();
}
box_ptr<Source> narFromPath(const StorePath & path) override
@@ -255,7 +259,7 @@ struct LegacySSHStore final : public Store
std::optional<StorePath> queryPathFromHashPart(const std::string & hashPart) override
{ unsupported("queryPathFromHashPart"); }
StorePath addToStore(
kj::Promise<Result<StorePath>> addToStore(
std::string_view name,
const Path & srcPath,
FileIngestionMethod method,
@@ -263,7 +267,7 @@ struct LegacySSHStore final : public Store
PathFilter & filter,
RepairFlag repair,
const StorePathSet & references) override
{ unsupported("addToStore"); }
try { unsupported("addToStore"); } catch (...) { return {result::current_exception()}; }
StorePath addTextToStore(
std::string_view name,
lix/libstore/local-store.cc
+5 -2
View File
@@ -1195,9 +1195,9 @@ bool LocalStore::realisationIsUntrusted(const Realisation & realisation)
return config_.requireSigs && !realisation.checkSignatures(getPublicKeys());
}
void LocalStore::addToStore(const ValidPathInfo & info, Source & source,
kj::Promise<Result<void>> LocalStore::addToStore(const ValidPathInfo & info, Source & 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));
@@ -1275,6 +1275,9 @@ void LocalStore::addToStore(const ValidPathInfo & info, Source & source,
registerValidPath(info);
}
}
co_return result::success();
} catch (...) {
co_return result::current_exception();
}
lix/libstore/local-store.hh
+1 -1
View File
@@ -204,7 +204,7 @@ public:
bool pathInfoIsUntrusted(const ValidPathInfo &) override;
bool realisationIsUntrusted(const Realisation & ) override;
void addToStore(const ValidPathInfo & info, Source & source,
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & source,
RepairFlag repair, CheckSigsFlag checkSigs) override;
StorePath addToStoreFromDump(Source & dump, std::string_view name,
lix/libstore/make-content-addressed.cc
+1 -1
View File
@@ -69,7 +69,7 @@ try {
info.narSize = sink.s.size();
StringSource source(rewritten);
dstStore.addToStore(info, source);
TRY_AWAIT(dstStore.addToStore(info, source));
remappings.insert_or_assign(std::move(path), std::move(info.path));
}
lix/libstore/remote-store.cc
+5 -2
View File
@@ -465,9 +465,9 @@ StorePath RemoteStore::addToStoreFromDump(Source & dump, std::string_view name,
}
void RemoteStore::addToStore(const ValidPathInfo & info, Source & source,
kj::Promise<Result<void>> RemoteStore::addToStore(const ValidPathInfo & info, Source & source,
RepairFlag repair, CheckSigsFlag checkSigs)
{
try {
auto conn(getConnection());
conn->to << WorkerProto::Op::AddToStoreNar
@@ -486,6 +486,9 @@ void RemoteStore::addToStore(const ValidPathInfo & info, Source & source,
} else {
conn.processStderr(0, &source);
}
co_return result::success();
} catch (...) {
co_return result::current_exception();
}
lix/libstore/remote-store.hh
+1 -1
View File
@@ -88,7 +88,7 @@ public:
StorePath addToStoreFromDump(Source & dump, std::string_view name,
FileIngestionMethod method = FileIngestionMethod::Recursive, HashType hashAlgo = HashType::SHA256, RepairFlag repair = NoRepair, const StorePathSet & references = StorePathSet()) override;
void addToStore(const ValidPathInfo & info, Source & nar,
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & nar,
RepairFlag repair, CheckSigsFlag checkSigs) override;
kj::Promise<Result<void>> addMultipleToStore(
lix/libstore/store-api.cc
+12 -9
View File
@@ -269,7 +269,7 @@ StorePath Store::computeStorePathForText(
}
StorePath Store::addToStore(
kj::Promise<Result<StorePath>> Store::addToStore(
std::string_view name,
const Path & _srcPath,
FileIngestionMethod method,
@@ -277,13 +277,15 @@ StorePath Store::addToStore(
PathFilter & filter,
RepairFlag repair,
const StorePathSet & references)
{
try {
Path srcPath(absPath(_srcPath));
auto source = GeneratorSource{
method == FileIngestionMethod::Recursive ? dumpPath(srcPath, filter).decay()
: readFileSource(srcPath)
};
return addToStoreFromDump(source, name, method, hashAlgo, repair, references);
co_return addToStoreFromDump(source, name, method, hashAlgo, repair, references);
} catch (...) {
co_return result::current_exception();
}
kj::Promise<Result<void>> Store::addMultipleToStore(
@@ -313,7 +315,7 @@ try {
processGraph<StorePath>("addMultipleToStore pool",
storePathsToAdd,
[&](const StorePath & path) {
[&](AsyncIoRoot & aio, const StorePath & path) {
auto & [info, _] = *infosMap.at(path);
@@ -329,7 +331,7 @@ try {
return info.references;
},
[&](const StorePath & path) {
[&](AsyncIoRoot & aio, const StorePath & path) {
checkInterrupt();
auto & [info_, source_] = *infosMap.at(path);
@@ -347,7 +349,7 @@ try {
MaintainCount<decltype(nrRunning)> mc(nrRunning);
showProgress();
try {
addToStore(info, *source, repair, checkSigs);
aio.blockOn(addToStore(info, *source, repair, checkSigs));
} catch (Error & e) {
nrFailed++;
if (!settings.keepGoing)
@@ -380,7 +382,7 @@ try {
WorkerProto::ReadConn {source, remoteVersion}
);
info.ultimate = false;
addToStore(info, source, repair, checkSigs);
TRY_AWAIT(addToStore(info, source, repair, checkSigs));
}
co_return result::success();
} catch (...) {
@@ -513,7 +515,7 @@ try {
if (!isValidPath(info.path)) {
auto source = GeneratorSource{dumpPath(srcPath)};
addToStore(info, source);
TRY_AWAIT(addToStore(info, source));
}
co_return info;
@@ -1091,6 +1093,7 @@ static std::string makeCopyPathMessage(
static constexpr unsigned PATH_COPY_BUFSIZE = 65536;
void copyStorePath(
AsyncIoRoot & aio,
Store & srcStore,
Store & dstStore,
const StorePath & storePath,
@@ -1147,7 +1150,7 @@ void copyStorePath(
}(act, info, srcStore, storePath)
};
dstStore.addToStore(*info, source, repair, checkSigs);
aio.blockOn(dstStore.addToStore(*info, source, repair, checkSigs));
}
lix/libstore/store-api.hh
+4 -2
View File
@@ -1,6 +1,7 @@
#pragma once
///@file
#include "lix/libutil/async.hh"
#include "lix/libutil/logging.hh"
#include "lix/libstore/nar-info.hh"
#include "lix/libstore/realisation.hh"
@@ -495,7 +496,7 @@ public:
/**
* Import a path into the store.
*/
virtual void addToStore(const ValidPathInfo & info, Source & narSource,
virtual kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & narSource,
RepairFlag repair = NoRepair, CheckSigsFlag checkSigs = CheckSigs) = 0;
/**
@@ -526,7 +527,7 @@ public:
* @param filter This function can be used to exclude files (see
* libutil/archive.hh).
*/
virtual StorePath addToStore(
virtual kj::Promise<Result<StorePath>> addToStore(
std::string_view name,
const Path & srcPath,
FileIngestionMethod method = FileIngestionMethod::Recursive,
@@ -907,6 +908,7 @@ protected:
* Copy a path from one store to another.
*/
void copyStorePath(
AsyncIoRoot & aio,
Store & srcStore,
Store & dstStore,
const StorePath & storePath,
lix/nix/add-to-store.cc
+1 -1
View File
@@ -55,7 +55,7 @@ struct CmdAddToStore : MixDryRun, StoreCommand
if (!dryRun) {
auto source = StringSource(sink.s);
store->addToStore(info, source);
aio().blockOn(store->addToStore(info, source));
}
logger->cout("%s", store->printStorePath(info.path));
perl/lib/Nix/Store.xs
+1 -1
View File
@@ -292,7 +292,7 @@ SV * addToStore(char * srcPath, int recursive, char * algo)
PPCODE:
try {
auto method = recursive ? FileIngestionMethod::Recursive : FileIngestionMethod::Flat;
auto path = store()->addToStore(std::string(baseNameOf(srcPath)), srcPath, method, parseHashType(algo));
auto path = aio().blockOn(store()->addToStore(std::string(baseNameOf(srcPath)), srcPath, method, parseHashType(algo)));
XPUSHs(sv_2mortal(newSVpv(store()->printStorePath(path).c_str(), 0)));
} catch (Error & e) {
croak("%s", e.what());