From de89c7f7c8847ff244a24d2f47bcf801a0c5a58d Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Sun, 15 Jun 2025 14:47:08 +0200 Subject: [PATCH] libstore: asyncify curl interface Change-Id: I3fc93016b8ac5e59d9062d4f4aead19ae051a680 --- lix/libfetchers/tarball.cc | 2 +- lix/libstore/builtins/fetchurl.cc | 47 ++++++++++++------- lix/libstore/filetransfer.cc | 35 +++++++++------ lix/libstore/filetransfer.hh | 9 ++-- lix/libstore/http-binary-cache-store.cc | 12 +++-- lix/nix/prefetch.cc | 2 +- lix/nix/upgrade-nix.cc | 2 +- tests/unit/libstore/filetransfer.cc | 60 +++++++++++++++++-------- 8 files changed, 112 insertions(+), 57 deletions(-) diff --git a/lix/libfetchers/tarball.cc b/lix/libfetchers/tarball.cc index 542c2aba1..3adc23ce9 100644 --- a/lix/libfetchers/tarball.cc +++ b/lix/libfetchers/tarball.cc @@ -85,7 +85,7 @@ try { FileTransferResult res; std::string data; try { - auto [meta, content] = getFileTransfer()->download(url, headers); + auto [meta, content] = TRY_AWAIT(getFileTransfer()->download(url, headers)); res = std::move(meta); data = content->drain(); } catch (FileTransferError & e) { diff --git a/lix/libstore/builtins/fetchurl.cc b/lix/libstore/builtins/fetchurl.cc index 8dd9bf7ab..d90fb0570 100644 --- a/lix/libstore/builtins/fetchurl.cc +++ b/lix/libstore/builtins/fetchurl.cc @@ -2,6 +2,7 @@ #include "lix/libstore/filetransfer.hh" #include "lix/libstore/store-api.hh" #include "lix/libutil/archive.hh" +#include "lix/libutil/async.hh" #include "lix/libutil/compression.hh" #include "lix/libutil/strings.hh" @@ -34,9 +35,11 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData, a forked process. */ auto fileTransfer = makeFileTransfer(); - auto fetch = [&](const std::string & url) { - - auto raw = fileTransfer->download(url).second; + // we also have to run the remainder of this function in a fresh thread so + // we can have an aio root. the existing root on the current thread is not + // safe to use because that would badly interfere with the parent process. + auto fetch = [&](AsyncIoRoot & aio, const std::string & url) { + auto raw = aio.blockOn(fileTransfer->download(url)).second; auto decompressor = makeDecompressionSource( unpack && mainUrl.ends_with(".xz") ? "xz" : "none", *raw); @@ -52,21 +55,33 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData, } }; - /* Try the hashed mirrors first. */ - if (getAttr("outputHashMode") == "flat") - for (auto hashedMirror : settings.hashedMirrors.get()) - try { - if (!hashedMirror.ends_with("/")) hashedMirror += '/'; - std::optional ht = parseHashTypeOpt(getAttr("outputHashAlgo")); - Hash h = newHashAllowEmpty(getAttr("outputHash"), ht); - fetch(hashedMirror + printHashType(h.type) + "/" + h.to_string(Base::Base16, false)); - return; - } catch (Error & e) { - debug(e.what()); + std::async(std::launch::async, [&] { + AsyncIoRoot aio; + + /* Try the hashed mirrors first. */ + if (getAttr("outputHashMode") == "flat") { + for (auto hashedMirror : settings.hashedMirrors.get()) { + try { + if (!hashedMirror.ends_with("/")) { + hashedMirror += '/'; + } + std::optional ht = parseHashTypeOpt(getAttr("outputHashAlgo")); + Hash h = newHashAllowEmpty(getAttr("outputHash"), ht); + fetch( + aio, + hashedMirror + printHashType(h.type) + "/" + + h.to_string(Base::Base16, false) + ); + return; + } catch (Error & e) { + debug(e.what()); + } } + } - /* Otherwise try the specified URL. */ - fetch(mainUrl); + /* Otherwise try the specified URL. */ + fetch(aio, mainUrl); + }).get(); } } diff --git a/lix/libstore/filetransfer.cc b/lix/libstore/filetransfer.cc index 55a2db2d5..7e944ca05 100644 --- a/lix/libstore/filetransfer.cc +++ b/lix/libstore/filetransfer.cc @@ -4,6 +4,7 @@ #include "lix/libstore/store-api.hh" #include "lix/libstore/s3.hh" #include "lix/libutil/regex.hh" +#include "lix/libutil/result.hh" #include "lix/libutil/signals.hh" #include "lix/libutil/strings.hh" #include "lix/libutil/thread-name.hh" @@ -692,9 +693,13 @@ struct curlFileTransfer : public FileTransfer } #endif - void upload(const std::string & uri, std::string data, const Headers & headers) override - { - enqueueFileTransfer(uri, headers, std::move(data), false); + kj::Promise> + upload(const std::string & uri, std::string data, const Headers & headers) override + try { + TRY_AWAIT(enqueueFileTransfer(uri, headers, std::move(data), false)); + co_return result::success(); + } catch (...) { + co_return result::current_exception(); } std::optional>> tryEagerTransfers( @@ -784,20 +789,22 @@ struct curlFileTransfer : public FileTransfer return std::nullopt; } - std::pair> enqueueFileTransfer( + kj::Promise>>> enqueueFileTransfer( const std::string & uri, const Headers & headers, std::optional data, bool noBody ) - { + try { if (auto eager = tryEagerTransfers(uri, headers, data, noBody)) { - return std::move(*eager); + co_return std::move(*eager); } auto source = make_box_ptr(*this, uri, headers, std::move(data), noBody); source->awaitData(); - return {source->metadata, std::move(source)}; + co_return {source->metadata, std::move(source)}; + } catch (...) { + co_return result::current_exception(); } struct TransferSource : Source @@ -972,21 +979,23 @@ struct curlFileTransfer : public FileTransfer } }; - bool exists(const std::string & uri, const Headers & headers) override - { + kj::Promise> exists(const std::string & uri, const Headers & headers) override + try { try { - enqueueFileTransfer(uri, headers, std::nullopt, true); - return true; + TRY_AWAIT(enqueueFileTransfer(uri, headers, std::nullopt, true)); + co_return true; } catch (FileTransferError & e) { /* S3 buckets return 403 if a file doesn't exist and the bucket is unlistable, so treat 403 as 404. */ if (e.error == FileTransfer::NotFound || e.error == FileTransfer::Forbidden) - return false; + co_return false; throw; } + } catch (...) { + co_return result::current_exception(); } - std::pair> + kj::Promise>>> download(const std::string & uri, const Headers & headers) override { return enqueueFileTransfer(uri, headers, std::nullopt, false); diff --git a/lix/libstore/filetransfer.hh b/lix/libstore/filetransfer.hh index 1891904b0..6f2812a8f 100644 --- a/lix/libstore/filetransfer.hh +++ b/lix/libstore/filetransfer.hh @@ -4,10 +4,12 @@ #include "lix/libutil/box_ptr.hh" #include "lix/libutil/ref.hh" #include "lix/libutil/logging.hh" +#include "lix/libutil/result.hh" #include "lix/libutil/serialise.hh" #include "lix/libutil/types.hh" #include "lix/libutil/config.hh" +#include #include #include @@ -40,7 +42,7 @@ struct FileTransfer /** * Upload some data. May throw a FileTransferError exception. */ - virtual void + virtual kj::Promise> upload(const std::string & uri, std::string data, const Headers & headers = {}) = 0; /** @@ -53,7 +55,8 @@ struct FileTransfer * * S3 objects are downloaded completely to answer this request. */ - virtual bool exists(const std::string & uri, const Headers & headers = {}) = 0; + virtual kj::Promise> + exists(const std::string & uri, const Headers & headers = {}) = 0; /** * Download a file, returning its contents through a source. Will not return @@ -62,7 +65,7 @@ struct FileTransfer * thrown by the returned source. The source will only throw errors detected * during the transfer itself (decompression errors, connection drops, etc). */ - virtual std::pair> + virtual kj::Promise>>> download(const std::string & uri, const Headers & headers = {}) = 0; enum Error { NotFound, Forbidden, Misc, Transient, Interrupted }; diff --git a/lix/libstore/http-binary-cache-store.cc b/lix/libstore/http-binary-cache-store.cc index 8d6f43f96..dc13ddf4c 100644 --- a/lix/libstore/http-binary-cache-store.cc +++ b/lix/libstore/http-binary-cache-store.cc @@ -125,7 +125,7 @@ protected: checkEnabled(); try { - co_return getFileTransfer()->exists(makeURI(path)); + co_return TRY_AWAIT(getFileTransfer()->exists(makeURI(path))); } catch (FileTransferError & e) { maybeDisable(); throw; @@ -142,7 +142,9 @@ protected: try { auto data = StreamToSourceAdapter(istream).drain(); try { - getFileTransfer()->upload(makeURI(path), std::move(data), {{"Content-Type", mimeType}}); + TRY_AWAIT(getFileTransfer()->upload( + makeURI(path), std::move(data), {{"Content-Type", mimeType}} + )); } catch (FileTransferError & e) { throw UploadToHTTP( "while uploading to HTTP binary cache at '%s': %s", cacheUri, e.msg() @@ -175,7 +177,9 @@ protected: { } }; - return {make_box_ptr(getFileTransfer()->download(makeURI(path)).second)}; + co_return make_box_ptr( + TRY_AWAIT(getFileTransfer()->download(makeURI(path))).second + ); } catch (FileTransferError & e) { if (e.error == FileTransfer::NotFound || e.error == FileTransfer::Forbidden) throw NoSuchBinaryCacheFile("file '%s' does not exist in binary cache '%s'", path, getUri()); @@ -183,7 +187,7 @@ protected: throw; } } catch (...) { - return {result::current_exception()}; + co_return result::current_exception(); } /** diff --git a/lix/nix/prefetch.cc b/lix/nix/prefetch.cc index 6d0dbdb79..afff1af2b 100644 --- a/lix/nix/prefetch.cc +++ b/lix/nix/prefetch.cc @@ -100,7 +100,7 @@ std::tuple prefetchFile( FdSink sink(fd.get()); - getFileTransfer()->download(url).second->drainInto(sink); + aio.blockOn(getFileTransfer()->download(url)).second->drainInto(sink); } /* Optionally unpack the file. */ diff --git a/lix/nix/upgrade-nix.cc b/lix/nix/upgrade-nix.cc index 4943b2eeb..cd483b400 100644 --- a/lix/nix/upgrade-nix.cc +++ b/lix/nix/upgrade-nix.cc @@ -287,7 +287,7 @@ struct CmdUpgradeNix : MixDryRun, EvalCommand Activity act(*logger, lvlInfo, actUnknown, "querying latest Nix version"); // FIXME: use nixos.org? - auto [res, content] = getFileTransfer()->download(storePathsUrl); + auto [res, content] = aio().blockOn(getFileTransfer()->download(storePathsUrl)); auto data = content->drain(); auto evaluator = std::make_unique(aio(), SearchPath{}, store); diff --git a/tests/unit/libstore/filetransfer.cc b/tests/unit/libstore/filetransfer.cc index 3ede42fc3..f2a17bbb8 100644 --- a/tests/unit/libstore/filetransfer.cc +++ b/tests/unit/libstore/filetransfer.cc @@ -1,4 +1,5 @@ #include "lix/libstore/filetransfer.hh" +#include "lix/libutil/async.hh" #include "lix/libutil/compression.hh" #include "lix/libutil/error.hh" #include "lix/libutil/signals.hh" @@ -206,12 +207,16 @@ TEST(FileTransfer, exceptionAbortsDownload) struct Done : BaseException {}; + AsyncIoRoot aio; auto ft = makeFileTransfer(); LambdaSink broken([](auto block) { throw Done(); }); auto [port, srv] = serveHTTP({{"200 ok", "", [](int) { return "foo"; }}}); - ASSERT_THROW(ft->download(fmt("http://[::1]:%d/index", port)).second->drainInto(broken), Done); + ASSERT_THROW( + aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drainInto(broken), + Done + ); // makeFileTransfer returns a ref<>, which cannot be cleared. since we also // can't default-construct it we'll have to overwrite it instead, but we'll @@ -228,19 +233,21 @@ TEST(FileTransfer, exceptionAbortsDownload) TEST(FileTransfer, exceptionAbortsRead) { auto [port, srv] = serveHTTP("200 ok", "content-length: 0\r\n", [] { return ""; }); + AsyncIoRoot aio; auto ft = makeFileTransfer(); char buf[10] = ""; - ASSERT_THROW(ft->download(fmt("http://[::1]:%d/index", port)).second->read(buf, 10), EndOfFile); + ASSERT_THROW( + aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->read(buf, 10), + EndOfFile + ); } TEST(FileTransfer, NOT_ON_DARWIN(reportsSetupErrors)) { auto [port, srv] = serveHTTP("404 not found", "", [] { return ""; }); + AsyncIoRoot aio; auto ft = makeFileTransfer(); - ASSERT_THROW( - ft->download(fmt("http://[::1]:%d/index", port)), - FileTransferError - ); + ASSERT_THROW(aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))), FileTransferError); } TEST(FileTransfer, NOT_ON_DARWIN(defersFailures)) @@ -253,8 +260,9 @@ TEST(FileTransfer, NOT_ON_DARWIN(defersFailures)) // might only do so once its internal buffer has already been filled.) return std::string(1024 * 1024, ' '); }); + AsyncIoRoot aio; auto ft = makeFileTransfer(0); - auto src = ft->download(fmt("http://[::1]:%d/index", port)).second; + auto src = aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second; ASSERT_THROW(src->drain(), FileTransferError); } @@ -264,10 +272,11 @@ TEST(FileTransfer, NOT_ON_DARWIN(handlesContentEncoding)) std::string compressed = compress("gzip", original); auto [port, srv] = serveHTTP("200 ok", "content-encoding: gzip\r\n", [&] { return compressed; }); + AsyncIoRoot aio; auto ft = makeFileTransfer(); StringSink sink; - ft->download(fmt("http://[::1]:%d/index", port)).second->drainInto(sink); + aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drainInto(sink); EXPECT_EQ(sink.s, original); } @@ -289,8 +298,9 @@ TEST(FileTransfer, usesIntermediateLinkHeaders) [] { return ""; }}, {"200 ok", "content-length: 1\r\n", [] { return "a"; }}, }); + AsyncIoRoot aio; auto ft = makeFileTransfer(0); - auto [result, _data] = ft->download(fmt("http://[::1]:%d/first", port)); + auto [result, _data] = aio.blockOn(ft->download(fmt("http://[::1]:%d/first", port))); ASSERT_EQ(result.immutableUrl, "http://foo"); } @@ -303,9 +313,10 @@ TEST(FileTransfer, stalledReaderDoesntBlockOthers) return round < 100 ? std::optional(std::string(1'000'000, ' ')) : std::nullopt; }}, }); + AsyncIoRoot aio; auto ft = makeFileTransfer(0); - auto [_result1, data1] = ft->download(fmt("http://[::1]:%d", port)); - auto [_result2, data2] = ft->download(fmt("http://[::1]:%d", port)); + auto [_result1, data1] = aio.blockOn(ft->download(fmt("http://[::1]:%d", port))); + auto [_result2, data2] = aio.blockOn(ft->download(fmt("http://[::1]:%d", port))); auto drop = [](Source & source, size_t size) { char buf[1000]; while (size > 0) { @@ -342,8 +353,9 @@ TEST(FileTransfer, retries) [] { return "b"; }, {"Range: bytes=1-"}}, }); + AsyncIoRoot aio; auto ft = makeFileTransfer(0); - auto [result, data] = ft->download(fmt("http://[::1]:%d", port)); + auto [result, data] = aio.blockOn(ft->download(fmt("http://[::1]:%d", port))); ASSERT_EQ(data->drain(), "ab"); } @@ -352,8 +364,9 @@ TEST(FileTransfer, doesntRetrySetupForever) auto [port, srv] = serveHTTP({ {"429 try again later", "content-length: 0\r\n", [] { return ""; }}, }); + AsyncIoRoot aio; auto ft = makeFileTransfer(0); - ASSERT_THROW(ft->download(fmt("http://[::1]:%d", port)), FileTransferError); + ASSERT_THROW(aio.blockOn(ft->download(fmt("http://[::1]:%d", port))), FileTransferError); } TEST(FileTransfer, doesntRetryTransferForever) @@ -374,12 +387,16 @@ TEST(FileTransfer, doesntRetryTransferForever) ); } auto [port, srv] = serveHTTP(replies); + AsyncIoRoot aio; auto ft = makeFileTransfer(0); - ASSERT_THROW(ft->download(fmt("http://[::1]:%d", port)).second->drain(), FileTransferError); + ASSERT_THROW( + aio.blockOn(ft->download(fmt("http://[::1]:%d", port))).second->drain(), FileTransferError + ); } TEST(FileTransfer, doesntRetryUploads) { + AsyncIoRoot aio; auto ft = makeFileTransfer(0); { @@ -387,14 +404,16 @@ TEST(FileTransfer, doesntRetryUploads) {"429 try again later", "", [] { return ""; }}, {"200 ok", "", [] { return ""; }}, }); - ASSERT_THROW(ft->upload(fmt("http://[::1]:%d", port), ""), FileTransferError); + ASSERT_THROW(aio.blockOn(ft->upload(fmt("http://[::1]:%d", port), "")), FileTransferError); } { auto [port, srv] = serveHTTP({ {"429 try again later", "", [] { return ""; }}, {"200 ok", "", [] { return ""; }}, }); - ASSERT_THROW(ft->upload(fmt("http://[::1]:%d", port), "foo"), FileTransferError); + ASSERT_THROW( + aio.blockOn(ft->upload(fmt("http://[::1]:%d", port), "foo")), FileTransferError + ); } } @@ -419,12 +438,16 @@ TEST(FileTransfer, DISABLED_interrupt) verbosity = lvlDebug; logger = new InterruptingLogger; + AsyncIoRoot aio; auto ft = makeFileTransfer(0); auto [port, srv] = serveHTTP({ {"200 ok", "content-length: 10\r\n", [] { return "0123456789"; }}, }); - ASSERT_THROW(ft->download(fmt("http://[::1]:%d/index", port)).second->drain(), FileTransferError); + ASSERT_THROW( + aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drain(), + FileTransferError + ); } TEST(FileTransfer, setupErrorsAreMetadata) @@ -432,8 +455,9 @@ TEST(FileTransfer, setupErrorsAreMetadata) auto [port, srv] = serveHTTP({ {"404 try again later", "content-length: 1\r\n", [] { return "X"; }}, }); + AsyncIoRoot aio; auto ft = makeFileTransfer(0); - ASSERT_THROW(ft->upload(fmt("http://[::1]:%d", port), ""), FileTransferError); + ASSERT_THROW(aio.blockOn(ft->upload(fmt("http://[::1]:%d", port), "")), FileTransferError); } }