From d08787e42e5a08f19af4b7e3c9630d703e60d1cb Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Sat, 18 May 2024 19:24:17 +0200 Subject: [PATCH] [WIP] remove remaining sinkToSource uses Change-Id: Id1ee0d2ad4a3774f4bbb960d76f0f76ac4f3eff9 --- src/libstore/builtins/fetchurl.cc | 22 ++++------ src/libstore/remote-store.cc | 22 +++++++--- src/libstore/store-api.cc | 73 ++++++++++++++++--------------- 3 files changed, 62 insertions(+), 55 deletions(-) diff --git a/src/libstore/builtins/fetchurl.cc b/src/libstore/builtins/fetchurl.cc index 3fe9e2c6c..ea1d892a2 100644 --- a/src/libstore/builtins/fetchurl.cc +++ b/src/libstore/builtins/fetchurl.cc @@ -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") { diff --git a/src/libstore/remote-store.cc b/src/libstore/remote-store.cc index bf2e60918..f8d715f69 100644 --- a/src/libstore/remote-store.cc +++ b/src/libstore/remote-store.cc @@ -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::write(*this, - WorkerProto::WriteConn {sink, remoteVersion}, + co_yield WorkerProto::Serialise::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( diff --git a/src/libstore/store-api.cc b/src/libstore/store-api.cc index 28a414555..fe7765fe9 100644 --- a/src/libstore/store-api.cc +++ b/src/libstore/store-api.cc @@ -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 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(source(srcStore, dstStore, missingPath, info)) + }); } dstStore.addMultipleToStore(pathsToCopy, act, repair, checkSigs);