From 1729c8ca3e6295b45520a855f43d67afc820700c Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Sun, 15 Jun 2025 14:47:08 +0200 Subject: [PATCH] libstore: asyncify curl return streams making stores and their users fully async requires all data streams to be async. the most notable data streams in common usage are curl first and remote stores second. curl is much more contained today and easier to asyncify (with the preparatory work we've done in the past commits) Change-Id: I2d6ff4687ee2b47e4efaa6714827b7283bed941d --- lix/libfetchers/tarball.cc | 2 +- lix/libstore/builtins/fetchurl.cc | 9 +- lix/libstore/filetransfer.cc | 208 ++++++++++++++++-------- lix/libstore/filetransfer.hh | 3 +- lix/libstore/http-binary-cache-store.cc | 14 +- lix/nix/prefetch.cc | 2 +- lix/nix/upgrade-nix.cc | 2 +- tests/unit/libstore/filetransfer.cc | 48 ++++-- 8 files changed, 184 insertions(+), 104 deletions(-) diff --git a/lix/libfetchers/tarball.cc b/lix/libfetchers/tarball.cc index 3adc23ce9..9019ee586 100644 --- a/lix/libfetchers/tarball.cc +++ b/lix/libfetchers/tarball.cc @@ -87,7 +87,7 @@ try { try { auto [meta, content] = TRY_AWAIT(getFileTransfer()->download(url, headers)); res = std::move(meta); - data = content->drain(); + data = TRY_AWAIT(content->drain()); } catch (FileTransferError & e) { if (cached) { warn("%s; using cached version", e.msg()); diff --git a/lix/libstore/builtins/fetchurl.cc b/lix/libstore/builtins/fetchurl.cc index d90fb0570..07d258c03 100644 --- a/lix/libstore/builtins/fetchurl.cc +++ b/lix/libstore/builtins/fetchurl.cc @@ -40,13 +40,14 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData, // 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); + auto decompressor = makeDecompressionStream( + unpack && mainUrl.ends_with(".xz") ? "xz" : "none", std::move(raw) + ); if (unpack) - restorePath(storePath, *decompressor); + aio.blockOn(restorePath(storePath, *decompressor)); else - writeFile(storePath, *decompressor); + aio.blockOn(writeFile(storePath, *decompressor)); auto executable = drv.env.find("executable"); if (executable != drv.env.end() && executable->second == "1") { diff --git a/lix/libstore/filetransfer.cc b/lix/libstore/filetransfer.cc index cc1443320..df7990753 100644 --- a/lix/libstore/filetransfer.cc +++ b/lix/libstore/filetransfer.cc @@ -1,4 +1,7 @@ #include "lix/libstore/filetransfer.hh" +#include "lix/libutil/async-io.hh" +#include "lix/libutil/async.hh" +#include "lix/libutil/error.hh" #include "lix/libutil/namespaces.hh" #include "lix/libstore/globals.hh" #include "lix/libstore/store-api.hh" @@ -12,7 +15,10 @@ #include #include +#include +#include #include +#include #if ENABLE_DTRACE #include "trace-probes.gen.hh" @@ -55,6 +61,22 @@ struct curlFileTransfer : public FileTransfer bool done = false; std::exception_ptr exc; std::string data; + std::optional>> downloadEvent; + + auto wait() + { + auto pfp = kj::newPromiseAndCrossThreadFulfiller(); + downloadEvent = std::move(pfp.fulfiller); + return std::move(pfp.promise); + } + + void signal() + { + if (downloadEvent) { + (*downloadEvent)->fulfill(); + downloadEvent.reset(); + } + } }; std::string uri; @@ -62,9 +84,8 @@ struct curlFileTransfer : public FileTransfer Activity act; std::unique_ptr uploadData; Sync downloadState; - std::condition_variable downloadEvent; bool headersDone = false, metadataReturned = false; - std::promise metadataPromise; + kj::Own>> metadataPromise; std::string statusMsg; uint64_t bodySize = 0; @@ -117,17 +138,23 @@ struct curlFileTransfer : public FileTransfer return uploadData ? "upload" : "download"; } - TransferItem(const std::string & uri, + TransferItem( + const std::string & uri, const Headers & headers, ActivityId parentAct, std::optional uploadData, bool noBody, - curl_off_t writtenToSink + curl_off_t writtenToSink, + kj::Own>> metadataPromise ) : uri(uri) - , act(*logger, lvlTalkative, actFileTransfer, - fmt(uploadData ? "uploading '%s'" : "downloading '%s'", uri), - {uri}, parentAct) + , act(*logger, + lvlTalkative, + actFileTransfer, + fmt(uploadData ? "uploading '%s'" : "downloading '%s'", uri), + {uri}, + parentAct) + , metadataPromise(std::move(metadataPromise)) , req(curl_easy_init()) { if (req == nullptr) { @@ -227,10 +254,10 @@ struct curlFileTransfer : public FileTransfer auto state = downloadState.lock(); assert(!state->done && !state->exc); if (!metadataReturned) { - metadataPromise.set_exception(ex); + metadataPromise->fulfill(ex); } state->exc = ex; - downloadEvent.notify_all(); + state->signal(); } template @@ -255,7 +282,7 @@ struct curlFileTransfer : public FileTransfer result.cached = status == 304; if (successfulStatuses.contains(status)) { - metadataPromise.set_value(result); + metadataPromise->fulfill(result); metadataReturned = true; } @@ -280,7 +307,7 @@ struct curlFileTransfer : public FileTransfer } state->data.append(static_cast(contents), realSize); - downloadEvent.notify_all(); + state->signal(); bodySize += realSize; return realSize; } catch (...) { @@ -373,8 +400,9 @@ struct curlFileTransfer : public FileTransfer else if (code == CURLE_OK && successfulStatuses.count(httpStatus)) { act.progress(bodySize, bodySize); - downloadState.lock()->done = true; - downloadEvent.notify_all(); + auto state = downloadState.lock(); + state->done = true; + state->signal(); } else { @@ -702,7 +730,7 @@ struct curlFileTransfer : public FileTransfer co_return result::current_exception(); } - std::optional>> tryEagerTransfers( + std::optional>> tryEagerTransfers( const std::string & uri, const Headers & headers, const std::optional & data, @@ -749,14 +777,25 @@ struct curlFileTransfer : public FileTransfer ); } if (S_ISDIR(st.st_mode)) { - return {{std::move(metadata), make_box_ptr("")}}; + return {{std::move(metadata), make_box_ptr("")}}; } - struct OwningFdSource : FdSource + struct OwningFdStream : AsyncInputStream { AutoCloseFD fd; - OwningFdSource(AutoCloseFD fd) : FdSource(fd.get()), fd(std::move(fd)) {} + OwningFdStream(AutoCloseFD fd) : fd(std::move(fd)) {} + kj::Promise> read(void * buffer, size_t size) override + { + // NOTE the synchronous implementation used to have a buffer for + // file data, but we cannot be bothered to treat this edge case. + if (const auto got = ::read(fd.get(), buffer, size); got >= 0) { + return {result::success(got)}; + } else { + return {result::failure(std::make_exception_ptr(SysError("reading file") + ))}; + } + } }; - return {{std::move(metadata), make_box_ptr(std::move(fd))}}; + return {{std::move(metadata), make_box_ptr(std::move(fd))}}; } } @@ -778,7 +817,15 @@ struct curlFileTransfer : public FileTransfer FileTransferResult res; if (!s3Res.data) throw FileTransferError(NotFound, "S3 object '%s' does not exist", uri); - return {{res, make_box_ptr(std::move(*s3Res.data))}}; + struct OwningStringStream : private std::string, AsyncStringInputStream + { + OwningStringStream(std::string data) + : std::string(std::move(data)) + , AsyncStringInputStream(*this) + { + } + }; + return {{ res, make_box_ptr(std::move(*s3Res.data)) }}; #else throw nix::Error( "cannot download '%s' because Lix is not built with S3 support", uri @@ -789,7 +836,8 @@ struct curlFileTransfer : public FileTransfer return std::nullopt; } - kj::Promise>>> enqueueFileTransfer( + kj::Promise>>> + enqueueFileTransfer( const std::string & uri, const Headers & headers, std::optional data, @@ -800,15 +848,15 @@ struct curlFileTransfer : public FileTransfer co_return std::move(*eager); } - auto source = make_box_ptr(*this, uri, headers, std::move(data), noBody); - source->init(); - source->awaitData(); + auto source = make_box_ptr(*this, uri, headers, std::move(data), noBody); + TRY_AWAIT(source->init()); + TRY_AWAIT(source->awaitData()); co_return {source->metadata, std::move(source)}; } catch (...) { co_return result::current_exception(); } - struct TransferSource : Source + struct TransferStream : AsyncInputStream { curlFileTransfer & parent; std::string uri; @@ -826,7 +874,7 @@ struct curlFileTransfer : public FileTransfer const size_t tries = fileTransferSettings.tries; curl_off_t totalReceived = 0; - TransferSource( + TransferStream( curlFileTransfer & parent, const std::string & uri, const Headers & headers, @@ -841,7 +889,7 @@ struct curlFileTransfer : public FileTransfer { } - ~TransferSource() + ~TransferStream() { // wake up the download thread if it's still going and have it abort try { @@ -853,22 +901,25 @@ struct curlFileTransfer : public FileTransfer } } - void init() - { + kj::Promise> init() + try { auto setup = [&] { return startTransfer(uri); }; - metadata = withRetries(setup, setup); + metadata = TRY_AWAIT(withRetries(setup, setup)); + co_return result::success(); + } catch (...) { + co_return result::current_exception(); } auto withRetries(auto && initial, auto && retry) -> decltype(initial()) - { + try { std::optional retryContext; while (true) { try { if (retryContext) { - prepareRetry(*retryContext); - return retry(); + TRY_AWAIT(prepareRetry(*retryContext)); + co_return TRY_AWAIT(retry()); } else { - return initial(); + co_return TRY_AWAIT(initial()); } } catch (FileTransferError & e) { // If this is a transient error, then maybe retry after a while. after any @@ -882,16 +933,23 @@ struct curlFileTransfer : public FileTransfer retryContext = e.what(); } } + } catch (...) { + co_return result::current_exception(); } - FileTransferResult startTransfer(const std::string & uri, curl_off_t offset = 0) - { + kj::Promise> + startTransfer(const std::string & uri, curl_off_t offset = 0) + try { attempt += 1; auto uploadData = data ? std::optional(std::string_view(*data)) : std::nullopt; - transfer = - std::make_shared(uri, headers, parentAct, uploadData, noBody, offset); + auto pfp = kj::newPromiseAndCrossThreadFulfiller>(); + transfer = std::make_shared( + uri, headers, parentAct, uploadData, noBody, offset, std::move(pfp.fulfiller) + ); parent.enqueueItem(transfer); - return transfer->metadataPromise.get_future().get(); + co_return TRY_AWAIT(pfp.promise); + } catch (...) { + co_return result::current_exception(); } void throwChangedTarget(std::string_view what, std::string_view from, std::string_view to) @@ -903,8 +961,8 @@ struct curlFileTransfer : public FileTransfer } } - void prepareRetry(const std::string & context) - { + kj::Promise> prepareRetry(const std::string & context) + try { thread_local std::minstd_rand random{std::random_device{}()}; std::uniform_real_distribution<> dist(0.0, 0.5); int ms = parent.baseRetryTimeMs * std::pow(2.0f, attempt - 1 + dist(random)); @@ -914,16 +972,19 @@ struct curlFileTransfer : public FileTransfer warn("%s; retrying in %d ms (attempt %d/%d)", context, ms, attempt, tries); } - std::this_thread::sleep_for(std::chrono::milliseconds(ms)); + co_await AIO().provider.getTimer().afterDelay(ms * kj::MILLISECONDS); + co_return result::success(); + } catch (...) { + co_return result::current_exception(); } - void restartTransfer() - { + kj::Promise> restartTransfer() + try { // use the effective URI of the previous transfer for retries. this avoids // some silent corruption if a redirect changes between starting and retry const auto & uri = metadata.effectiveUri.empty() ? this->uri : metadata.effectiveUri; - auto newMeta = startTransfer(uri, totalReceived); + auto newMeta = TRY_AWAIT(startTransfer(uri, totalReceived)); throwChangedTarget("final destination", metadata.effectiveUri, newMeta.effectiveUri); throwChangedTarget("ETag", metadata.etag, newMeta.etag); throwChangedTarget( @@ -931,13 +992,23 @@ struct curlFileTransfer : public FileTransfer metadata.immutableUrl.value_or(""), newMeta.immutableUrl.value_or("") ); + co_return result::success(); + } catch (...) { + co_return result::current_exception(); } - bool waitForData() - { + kj::Promise> waitForData() + try { /* Grab data if available, otherwise wait for the download thread to wake us up. */ + std::optional> signal; + while (buffered.empty()) { + if (signal) { + co_await *signal; + signal.reset(); + } + auto state(transfer->downloadState.lock()); if (!state->data.empty()) { @@ -948,46 +1019,51 @@ struct curlFileTransfer : public FileTransfer } else if (state->exc) { std::rethrow_exception(state->exc); } else if (state->done) { - return false; + co_return false; } else { parent.unpause(transfer); - state.wait(transfer->downloadEvent); + signal = state->wait(); } } - return true; + co_return true; + } catch (...) { + co_return result::current_exception(); } - bool restartAndWaitForData() - { - restartTransfer(); - return waitForData(); + kj::Promise> restartAndWaitForData() + try { + TRY_AWAIT(restartTransfer()); + co_return TRY_AWAIT(waitForData()); + } catch (...) { + co_return result::current_exception(); } - bool awaitData() - { - return withRetries( - [&] { return waitForData(); }, [&] { return restartAndWaitForData(); } + kj::Promise> awaitData() + try { + co_return TRY_AWAIT( + withRetries([&] { return waitForData(); }, [&] { return restartAndWaitForData(); }) ); + } catch (...) { + co_return result::current_exception(); } - size_t read(char * data, size_t len) override - { + kj::Promise> read(void * buffer, size_t len) override + try { TRACE(LIX_STORE_FILETRANSFER_READ(uri.c_str(), len)); size_t total = 0; - while (total < len && awaitData()) { + auto data = static_cast(buffer); + while (total < len && TRY_AWAIT(awaitData())) { const auto available = std::min(len - total, buffered.size()); memcpy(data + total, buffered.data(), available); buffered.remove_prefix(available); total += available; } - if (total == 0) { - throw EndOfFile("transfer finished"); - } - - return total; + co_return total; + } catch (...) { + co_return result::current_exception(); } }; @@ -1007,7 +1083,7 @@ struct curlFileTransfer : public FileTransfer co_return result::current_exception(); } - kj::Promise>>> + 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 6f2812a8f..42644f7aa 100644 --- a/lix/libstore/filetransfer.hh +++ b/lix/libstore/filetransfer.hh @@ -1,6 +1,7 @@ #pragma once ///@file +#include "lix/libutil/async-io.hh" #include "lix/libutil/box_ptr.hh" #include "lix/libutil/ref.hh" #include "lix/libutil/logging.hh" @@ -65,7 +66,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 kj::Promise>>> + 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 dc13ddf4c..a7c6f0634 100644 --- a/lix/libstore/http-binary-cache-store.cc +++ b/lix/libstore/http-binary-cache-store.cc @@ -167,19 +167,7 @@ protected: try { checkEnabled(); try { - struct HttpFile : AsyncSourceInputStream - { - box_ptr source; - - HttpFile(box_ptr source) - : AsyncSourceInputStream(*source) - , source(std::move(source)) - { - } - }; - co_return make_box_ptr( - TRY_AWAIT(getFileTransfer()->download(makeURI(path))).second - ); + co_return 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()); diff --git a/lix/nix/prefetch.cc b/lix/nix/prefetch.cc index afff1af2b..f8a365234 100644 --- a/lix/nix/prefetch.cc +++ b/lix/nix/prefetch.cc @@ -100,7 +100,7 @@ std::tuple prefetchFile( FdSink sink(fd.get()); - aio.blockOn(getFileTransfer()->download(url)).second->drainInto(sink); + aio.blockOn(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 cd483b400..35a1a964b 100644 --- a/lix/nix/upgrade-nix.cc +++ b/lix/nix/upgrade-nix.cc @@ -288,7 +288,7 @@ struct CmdUpgradeNix : MixDryRun, EvalCommand // FIXME: use nixos.org? auto [res, content] = aio().blockOn(getFileTransfer()->download(storePathsUrl)); - auto data = content->drain(); + auto data = aio().blockOn(content->drain()); auto evaluator = std::make_unique(aio(), SearchPath{}, store); auto state = evaluator->begin(aio()); diff --git a/tests/unit/libstore/filetransfer.cc b/tests/unit/libstore/filetransfer.cc index f2a17bbb8..b12baa2be 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-io.hh" #include "lix/libutil/async.hh" #include "lix/libutil/compression.hh" #include "lix/libutil/error.hh" @@ -214,7 +215,9 @@ TEST(FileTransfer, exceptionAbortsDownload) auto [port, srv] = serveHTTP({{"200 ok", "", [](int) { return "foo"; }}}); ASSERT_THROW( - aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drainInto(broken), + aio.blockOn( + aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drainInto(broken) + ), Done ); @@ -236,9 +239,11 @@ TEST(FileTransfer, exceptionAbortsRead) AsyncIoRoot aio; auto ft = makeFileTransfer(); char buf[10] = ""; - ASSERT_THROW( - aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->read(buf, 10), - EndOfFile + ASSERT_EQ( + aio.blockOn( + aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->read(buf, 10) + ), + 0 ); } @@ -263,7 +268,7 @@ TEST(FileTransfer, NOT_ON_DARWIN(defersFailures)) AsyncIoRoot aio; auto ft = makeFileTransfer(0); auto src = aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second; - ASSERT_THROW(src->drain(), FileTransferError); + ASSERT_THROW(aio.blockOn(src->drain()), FileTransferError); } TEST(FileTransfer, NOT_ON_DARWIN(handlesContentEncoding)) @@ -276,7 +281,9 @@ TEST(FileTransfer, NOT_ON_DARWIN(handlesContentEncoding)) auto ft = makeFileTransfer(); StringSink sink; - aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drainInto(sink); + aio.blockOn( + aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drainInto(sink) + ); EXPECT_EQ(sink.s, original); } @@ -317,23 +324,29 @@ TEST(FileTransfer, stalledReaderDoesntBlockOthers) auto ft = makeFileTransfer(0); 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) { + auto drop = [&](AsyncInputStream & source, size_t size) { char buf[1000]; + size_t dropped = 0; while (size > 0) { auto round = std::min(size, sizeof(buf)); - source(buf, round); + round = aio.blockOn(source.read(buf, round)); + if (round == 0) { + break; + } size -= round; + dropped += round; } + return dropped; }; // read 10M of each of the 100M, then the rest. neither reader should // block the other, nor should it take that long to copy 200MB total. - drop(*data1, 10'000'000); - drop(*data2, 10'000'000); - drop(*data1, 90'000'000); - drop(*data2, 90'000'000); + ASSERT_EQ(drop(*data1, 10'000'000), 10'000'000); + ASSERT_EQ(drop(*data2, 10'000'000), 10'000'000); + ASSERT_EQ(drop(*data1, 90'000'000), 90'000'000); + ASSERT_EQ(drop(*data2, 90'000'000), 90'000'000); - ASSERT_THROW(drop(*data1, 1), EndOfFile); - ASSERT_THROW(drop(*data2, 1), EndOfFile); + ASSERT_EQ(drop(*data1, 1), 0); + ASSERT_EQ(drop(*data2, 1), 0); } TEST(FileTransfer, retries) @@ -356,7 +369,7 @@ TEST(FileTransfer, retries) AsyncIoRoot aio; auto ft = makeFileTransfer(0); auto [result, data] = aio.blockOn(ft->download(fmt("http://[::1]:%d", port))); - ASSERT_EQ(data->drain(), "ab"); + ASSERT_EQ(aio.blockOn(data->drain()), "ab"); } TEST(FileTransfer, doesntRetrySetupForever) @@ -390,7 +403,8 @@ TEST(FileTransfer, doesntRetryTransferForever) AsyncIoRoot aio; auto ft = makeFileTransfer(0); ASSERT_THROW( - aio.blockOn(ft->download(fmt("http://[::1]:%d", port))).second->drain(), FileTransferError + aio.blockOn(aio.blockOn(ft->download(fmt("http://[::1]:%d", port))).second->drain()), + FileTransferError ); } @@ -445,7 +459,7 @@ TEST(FileTransfer, DISABLED_interrupt) }); ASSERT_THROW( - aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drain(), + aio.blockOn(aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drain()), FileTransferError ); }