[WIP] remove remaining sinkToSource uses
Change-Id: Id1ee0d2ad4a3774f4bbb960d76f0f76ac4f3eff9
This commit is contained in:
@@ -32,23 +32,19 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData)
|
||||
|
||||
auto fetch = [&](const std::string & url) {
|
||||
|
||||
auto source = sinkToSource([&](Sink & sink) {
|
||||
/* No need to do TLS verification, because we check the hash of
|
||||
the result anyway. */
|
||||
FileTransferRequest request(url);
|
||||
request.verifyTLS = false;
|
||||
|
||||
/* No need to do TLS verification, because we check the hash of
|
||||
the result anyway. */
|
||||
FileTransferRequest request(url);
|
||||
request.verifyTLS = false;
|
||||
|
||||
auto raw = fileTransfer->download(std::move(request));
|
||||
auto decompressor = makeDecompressionSource(
|
||||
unpack && mainUrl.ends_with(".xz") ? "xz" : "none", *raw);
|
||||
decompressor->drainInto(sink);
|
||||
});
|
||||
auto raw = fileTransfer->download(std::move(request));
|
||||
auto decompressor = makeDecompressionSource(
|
||||
unpack && mainUrl.ends_with(".xz") ? "xz" : "none", *raw);
|
||||
|
||||
if (unpack)
|
||||
restorePath(storePath, *source);
|
||||
restorePath(storePath, *decompressor);
|
||||
else
|
||||
writeFile(storePath, *source);
|
||||
writeFile(storePath, *decompressor);
|
||||
|
||||
auto executable = drv.env.find("executable");
|
||||
if (executable != drv.env.end() && executable->second == "1") {
|
||||
|
||||
@@ -485,17 +485,25 @@ void RemoteStore::addMultipleToStore(
|
||||
{
|
||||
auto remoteVersion = getProtocol();
|
||||
|
||||
auto source = sinkToSource([&](Sink & sink) {
|
||||
sink << pathsToCopy.size();
|
||||
GeneratorSource source{[](auto self, auto & pathsToCopy, auto remoteVersion) -> WireFormatGenerator {
|
||||
NullSink null; // TODO remove .to from WriteConn instead
|
||||
co_yield pathsToCopy.size();
|
||||
for (auto & [pathInfo, pathSource] : pathsToCopy) {
|
||||
sink << WorkerProto::Serialise<ValidPathInfo>::write(*this,
|
||||
WorkerProto::WriteConn {sink, remoteVersion},
|
||||
co_yield WorkerProto::Serialise<ValidPathInfo>::write(*self,
|
||||
WorkerProto::WriteConn {null, remoteVersion},
|
||||
pathInfo);
|
||||
pathSource->drainInto(sink);
|
||||
try {
|
||||
char buf[65536];
|
||||
while (true) {
|
||||
const auto read = pathSource->read(buf, sizeof(buf));
|
||||
co_yield std::span{buf, read};
|
||||
}
|
||||
} catch (EndOfFile &) {
|
||||
}
|
||||
}
|
||||
});
|
||||
}(this, pathsToCopy, remoteVersion)};
|
||||
|
||||
addMultipleToStore(*source, repair, checkSigs);
|
||||
addMultipleToStore(source, repair, checkSigs);
|
||||
}
|
||||
|
||||
void RemoteStore::addMultipleToStore(
|
||||
|
||||
+38
-35
@@ -335,9 +335,9 @@ void Store::addMultipleToStore(
|
||||
info.ultimate = false;
|
||||
|
||||
/* Make sure that the Source object is destroyed when
|
||||
we're done. In particular, a SinkToSource object must
|
||||
be destroyed to ensure that the destructors on its
|
||||
stack frame are run; this includes
|
||||
we're done. In particular, a coroutine object must
|
||||
be destroyed to ensure that the destructors in its
|
||||
state are run; this includes
|
||||
LegacySSHStore::narFromPath()'s connection lock. */
|
||||
auto source = std::move(source_);
|
||||
|
||||
@@ -1059,16 +1059,17 @@ void copyStorePath(
|
||||
info = info2;
|
||||
}
|
||||
|
||||
auto source = sinkToSource([&](Sink & sink) {
|
||||
LambdaSink progressSink([&, total = 0ULL](std::string_view data) mutable {
|
||||
total += data.size();
|
||||
GeneratorSource source{[](auto & act, auto & info, auto & srcStore, auto & storePath) -> WireFormatGenerator {
|
||||
auto nar = srcStore.narFromPath(storePath);
|
||||
uint64_t total = 0;
|
||||
while (auto data = nar.next()) {
|
||||
total += data->size();
|
||||
act.progress(total, info->narSize);
|
||||
});
|
||||
TeeSink tee { sink, progressSink };
|
||||
tee << srcStore.narFromPath(storePath);
|
||||
});
|
||||
co_yield *data;
|
||||
}
|
||||
}(act, info, srcStore, storePath)};
|
||||
|
||||
dstStore.addToStore(*info, *source, repair, checkSigs);
|
||||
dstStore.addToStore(*info, source, repair, checkSigs);
|
||||
}
|
||||
|
||||
|
||||
@@ -1180,31 +1181,33 @@ std::map<StorePath, StorePath> copyPaths(
|
||||
ValidPathInfo infoForDst = *info;
|
||||
infoForDst.path = storePathForDst;
|
||||
|
||||
auto source =
|
||||
sinkToSource([&srcStore, &dstStore, missingPath = missingPath, info = std::move(info)](Sink & sink) {
|
||||
// We can reasonably assume that the copy will happen whenever we
|
||||
// read the path, so log something about that at that point
|
||||
auto srcUri = srcStore.getUri();
|
||||
auto dstUri = dstStore.getUri();
|
||||
auto storePathS = srcStore.printStorePath(missingPath);
|
||||
Activity act(
|
||||
*logger,
|
||||
lvlInfo,
|
||||
actCopyPath,
|
||||
makeCopyPathMessage(srcUri, dstUri, storePathS),
|
||||
{storePathS, srcUri, dstUri}
|
||||
);
|
||||
PushActivity pact(act.id);
|
||||
auto source = [](auto & srcStore, auto & dstStore, auto missingPath, auto info
|
||||
) -> WireFormatGenerator {
|
||||
// We can reasonably assume that the copy will happen whenever we
|
||||
// read the path, so log something about that at that point
|
||||
auto srcUri = srcStore.getUri();
|
||||
auto dstUri = dstStore.getUri();
|
||||
auto storePathS = srcStore.printStorePath(missingPath);
|
||||
Activity act(
|
||||
*logger,
|
||||
lvlInfo,
|
||||
actCopyPath,
|
||||
makeCopyPathMessage(srcUri, dstUri, storePathS),
|
||||
{storePathS, srcUri, dstUri}
|
||||
);
|
||||
PushActivity pact(act.id);
|
||||
|
||||
LambdaSink progressSink([&, total = 0ULL](std::string_view data) mutable {
|
||||
total += data.size();
|
||||
act.progress(total, info->narSize);
|
||||
});
|
||||
TeeSink tee{sink, progressSink};
|
||||
|
||||
tee << srcStore.narFromPath(missingPath);
|
||||
});
|
||||
pathsToCopy.push_back(std::pair{infoForDst, std::move(source)});
|
||||
auto nar = srcStore.narFromPath(missingPath);
|
||||
uint64_t total = 0;
|
||||
while (auto data = nar.next()) {
|
||||
total += data->size();
|
||||
act.progress(total, info->narSize);
|
||||
co_yield *data;
|
||||
}
|
||||
};
|
||||
pathsToCopy.push_back(std::pair{
|
||||
infoForDst, std::make_unique<GeneratorSource>(source(srcStore, dstStore, missingPath, info))
|
||||
});
|
||||
}
|
||||
|
||||
dstStore.addMultipleToStore(pathsToCopy, act, repair, checkSigs);
|
||||
|
||||
Reference in New Issue
Block a user