From c34e96f7e0d53894f302134b2f90f83b20ffd22a Mon Sep 17 00:00:00 2001 From: Silvan Mosberger Date: Thu, 2 Apr 2020 05:03:58 +0200 Subject: Make function arguments retain position info This allows querying the location of function arguments. E.g. builtins.unsafeGetAttrPos "x" (builtins.functionArgs ({ x }: null)) => { column = 57; file = "/home/infinisil/src/nix/inst/test.nix"; line = 1; } --- src/libexpr/nixexpr.hh | 3 ++- src/libexpr/parser.y | 4 ++-- src/libexpr/primops.cc | 7 +++++-- 3 files changed, 9 insertions(+), 5 deletions(-) (limited to 'src') diff --git a/src/libexpr/nixexpr.hh b/src/libexpr/nixexpr.hh index f7e9105a4..8c96de37c 100644 --- a/src/libexpr/nixexpr.hh +++ b/src/libexpr/nixexpr.hh @@ -209,9 +209,10 @@ struct ExprList : Expr struct Formal { + Pos pos; Symbol name; Expr * def; - Formal(const Symbol & name, Expr * def) : name(name), def(def) { }; + Formal(const Pos & pos, const Symbol & name, Expr * def) : pos(pos), name(name), def(def) { }; }; struct Formals diff --git a/src/libexpr/parser.y b/src/libexpr/parser.y index 9c769e803..08cec96dc 100644 --- a/src/libexpr/parser.y +++ b/src/libexpr/parser.y @@ -531,8 +531,8 @@ formals ; formal - : ID { $$ = new Formal(data->symbols.create($1), 0); } - | ID '?' expr { $$ = new Formal(data->symbols.create($1), $3); } + : ID { $$ = new Formal(CUR_POS, data->symbols.create($1), 0); } + | ID '?' expr { $$ = new Formal(CUR_POS, data->symbols.create($1), $3); } ; %% diff --git a/src/libexpr/primops.cc b/src/libexpr/primops.cc index 8de234951..81c378dff 100644 --- a/src/libexpr/primops.cc +++ b/src/libexpr/primops.cc @@ -1354,9 +1354,12 @@ static void prim_functionArgs(EvalState & state, const Pos & pos, Value * * args } state.mkAttrs(v, args[0]->lambda.fun->formals->formals.size()); - for (auto & i : args[0]->lambda.fun->formals->formals) + for (auto & i : args[0]->lambda.fun->formals->formals) { // !!! should optimise booleans (allocate only once) - mkBool(*state.allocAttr(v, i.name), i.def); + Value * value = state.allocValue(); + v.attrs->push_back(Attr(i.name, value, &i.pos)); + mkBool(*value, i.def); + } v.attrs->sort(); } -- cgit v1.2.3 From 9d04b5da17898fd564308423bc7d10fc929efe18 Mon Sep 17 00:00:00 2001 From: mlatus Date: Tue, 7 Apr 2020 20:29:40 +0800 Subject: `nix run` using $SHELL as default command --- src/nix/run.cc | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'src') diff --git a/src/nix/run.cc b/src/nix/run.cc index 8e30264c0..5334531fd 100644 --- a/src/nix/run.cc +++ b/src/nix/run.cc @@ -59,14 +59,14 @@ struct RunCommon : virtual Command struct CmdRun : InstallablesCommand, RunCommon, MixEnvironment { - std::vector command = { "bash" }; + std::vector command = { getEnv("SHELL").value_or("bash") }; CmdRun() { mkFlag() .longName("command") .shortName('c') - .description("command and arguments to be executed; defaults to 'bash'") + .description("command and arguments to be executed; defaults to '$SHELL'") .labels({"command", "args"}) .arity(ArityAny) .handler([&](std::vector ss) { -- cgit v1.2.3 From 1ab8d6ac1861e9405ae34af3deb681020c03e82d Mon Sep 17 00:00:00 2001 From: Eelco Dolstra Date: Wed, 8 Apr 2020 15:18:41 +0200 Subject: Downloader: Only write data to the sink on a 200 response Hopefully fixes #3278. --- src/libstore/download.cc | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) (limited to 'src') diff --git a/src/libstore/download.cc b/src/libstore/download.cc index af69699a8..bc875f653 100644 --- a/src/libstore/download.cc +++ b/src/libstore/download.cc @@ -83,8 +83,15 @@ struct CurlDownloader : public Downloader , callback(std::move(callback)) , finalSink([this](const unsigned char * data, size_t len) { if (this->request.dataCallback) { - writtenToSink += len; - this->request.dataCallback((char *) data, len); + long httpStatus = 0; + curl_easy_getinfo(req, CURLINFO_RESPONSE_CODE, &httpStatus); + + /* Only write data to the sink if this is a + successful response. */ + if (httpStatus == 0 || httpStatus == 200 || httpStatus == 201 || httpStatus == 206) { + writtenToSink += len; + this->request.dataCallback((char *) data, len); + } } else this->result.data->append((char *) data, len); }) -- cgit v1.2.3 From bf81b3155998152582abdad1550277cf8ca32e92 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Domen=20Ko=C5=BEar?= Date: Wed, 8 Apr 2020 18:27:10 +0200 Subject: build.cc: improve message if home directory exists --- src/libstore/build.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'src') diff --git a/src/libstore/build.cc b/src/libstore/build.cc index 527d7ac42..632982ecd 100644 --- a/src/libstore/build.cc +++ b/src/libstore/build.cc @@ -2161,7 +2161,7 @@ void DerivationGoal::startBuilder() if (needsHashRewrite()) { if (pathExists(homeDir)) - throw Error(format("directory '%1%' exists; please remove it") % homeDir); + throw Error(format("home directory '%1%' exists; please remove it to assure purity of builds without sandboxing") % homeDir); /* We're not doing a chroot build, but we have some valid output paths. Since we can't just overwrite or delete -- cgit v1.2.3 From 65ef57e0cbded52652510355304303a7b71586af Mon Sep 17 00:00:00 2001 From: Nikola Knezevic Date: Mon, 6 Apr 2020 13:30:45 +0200 Subject: DownloadRequest -> DataTransferRequest --- src/libstore/builtins/fetchurl.cc | 2 +- src/libstore/download.cc | 13 ++++++------- src/libstore/download.hh | 12 ++++++------ src/libstore/http-binary-cache-store.cc | 8 ++++---- src/nix-prefetch-url/nix-prefetch-url.cc | 2 +- src/nix/upgrade-nix.cc | 2 +- 6 files changed, 19 insertions(+), 20 deletions(-) (limited to 'src') diff --git a/src/libstore/builtins/fetchurl.cc b/src/libstore/builtins/fetchurl.cc index f6ae5d2e6..a939f040f 100644 --- a/src/libstore/builtins/fetchurl.cc +++ b/src/libstore/builtins/fetchurl.cc @@ -36,7 +36,7 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData) /* No need to do TLS verification, because we check the hash of the result anyway. */ - DownloadRequest request(url); + DataTransferRequest request(url); request.verifyTLS = false; request.decompress = false; diff --git a/src/libstore/download.cc b/src/libstore/download.cc index bc875f653..77f3c527a 100644 --- a/src/libstore/download.cc +++ b/src/libstore/download.cc @@ -49,7 +49,7 @@ struct CurlDownloader : public Downloader struct DownloadItem : public std::enable_shared_from_this { CurlDownloader & downloader; - DownloadRequest request; + DataTransferRequest request; DownloadResult result; Activity act; bool done = false; // whether either the success or failure function has been called @@ -73,7 +73,7 @@ struct CurlDownloader : public Downloader curl_off_t writtenToSink = 0; DownloadItem(CurlDownloader & downloader, - const DownloadRequest & request, + const DataTransferRequest & request, Callback && callback) : downloader(downloader) , request(request) @@ -641,7 +641,7 @@ struct CurlDownloader : public Downloader } #endif - void enqueueDownload(const DownloadRequest & request, + void enqueueDownload(const DataTransferRequest & request, Callback callback) override { /* Ugly hack to support s3:// URIs. */ @@ -687,7 +687,7 @@ ref makeDownloader() return make_ref(); } -std::future Downloader::enqueueDownload(const DownloadRequest & request) +std::future Downloader::enqueueDownload(const DataTransferRequest & request) { auto promise = std::make_shared>(); enqueueDownload(request, @@ -701,12 +701,12 @@ std::future Downloader::enqueueDownload(const DownloadRequest & return promise->get_future(); } -DownloadResult Downloader::download(const DownloadRequest & request) +DownloadResult Downloader::download(const DataTransferRequest & request) { return enqueueDownload(request).get(); } -void Downloader::download(DownloadRequest && request, Sink & sink) +void Downloader::download(DataTransferRequest && request, Sink & sink) { /* Note: we can't call 'sink' via request.dataCallback, because that would cause the sink to execute on the downloader @@ -801,7 +801,6 @@ void Downloader::download(DownloadRequest && request, Sink & sink) } } - bool isUri(const string & s) { if (s.compare(0, 8, "channel:") == 0) return true; diff --git a/src/libstore/download.hh b/src/libstore/download.hh index 28c8a9162..89d3fce7f 100644 --- a/src/libstore/download.hh +++ b/src/libstore/download.hh @@ -33,7 +33,7 @@ struct DownloadSettings : Config extern DownloadSettings downloadSettings; -struct DownloadRequest +struct DataTransferRequest { std::string uri; std::string expectedETag; @@ -47,7 +47,7 @@ struct DownloadRequest std::string mimeType; std::function dataCallback; - DownloadRequest(const std::string & uri) + DataTransferRequest(const std::string & uri) : uri(uri), parentAct(getCurActivity()) { } std::string verb() @@ -74,17 +74,17 @@ struct Downloader /* Enqueue a download request, returning a future to the result of the download. The future may throw a DownloadError exception. */ - virtual void enqueueDownload(const DownloadRequest & request, + virtual void enqueueDownload(const DataTransferRequest & request, Callback callback) = 0; - std::future enqueueDownload(const DownloadRequest & request); + std::future enqueueDownload(const DataTransferRequest & request); /* Synchronously download a file. */ - DownloadResult download(const DownloadRequest & request); + DownloadResult download(const DataTransferRequest & request); /* Download a file, writing its data to a sink. The sink will be invoked on the thread of the caller. */ - void download(DownloadRequest && request, Sink & sink); + void download(DataTransferRequest && request, Sink & sink); enum Error { NotFound, Forbidden, Misc, Transient, Interrupted }; }; diff --git a/src/libstore/http-binary-cache-store.cc b/src/libstore/http-binary-cache-store.cc index 011794c62..e152e7524 100644 --- a/src/libstore/http-binary-cache-store.cc +++ b/src/libstore/http-binary-cache-store.cc @@ -85,7 +85,7 @@ protected: checkEnabled(); try { - DownloadRequest request(cacheUri + "/" + path); + DataTransferRequest request(cacheUri + "/" + path); request.head = true; getDownloader()->download(request); return true; @@ -103,7 +103,7 @@ protected: const std::string & data, const std::string & mimeType) override { - auto req = DownloadRequest(cacheUri + "/" + path); + auto req = DataTransferRequest(cacheUri + "/" + path); req.data = std::make_shared(data); // FIXME: inefficient req.mimeType = mimeType; try { @@ -113,9 +113,9 @@ protected: } } - DownloadRequest makeRequest(const std::string & path) + DataTransferRequest makeRequest(const std::string & path) { - DownloadRequest request(cacheUri + "/" + path); + DataTransferRequest request(cacheUri + "/" + path); return request; } diff --git a/src/nix-prefetch-url/nix-prefetch-url.cc b/src/nix-prefetch-url/nix-prefetch-url.cc index 2b9254659..bfd65da98 100644 --- a/src/nix-prefetch-url/nix-prefetch-url.cc +++ b/src/nix-prefetch-url/nix-prefetch-url.cc @@ -180,7 +180,7 @@ static int _main(int argc, char * * argv) FdSink sink(fd.get()); - DownloadRequest req(actualUri); + DataTransferRequest req(actualUri); req.decompress = false; getDownloader()->download(std::move(req), sink); } diff --git a/src/nix/upgrade-nix.cc b/src/nix/upgrade-nix.cc index c05c29517..aeaf7b09c 100644 --- a/src/nix/upgrade-nix.cc +++ b/src/nix/upgrade-nix.cc @@ -138,7 +138,7 @@ struct CmdUpgradeNix : MixDryRun, StoreCommand Activity act(*logger, lvlInfo, actUnknown, "querying latest Nix version"); // FIXME: use nixos.org? - auto req = DownloadRequest(storePathsUrl); + auto req = DataTransferRequest(storePathsUrl); auto res = getDownloader()->download(req); auto state = std::make_unique(Strings(), store); -- cgit v1.2.3 From 741e9012d3fa6f0fb41a4a5662c6b8b38ecfaa1f Mon Sep 17 00:00:00 2001 From: Nikola Knezevic Date: Mon, 6 Apr 2020 15:10:58 +0200 Subject: Rename src/lib/download.* to src/lib/datatransfer.* --- src/libexpr/common-eval-args.cc | 2 +- src/libexpr/eval.cc | 2 +- src/libexpr/parser.y | 2 +- src/libstore/build.cc | 2 +- src/libstore/builtins/fetchurl.cc | 2 +- src/libstore/datatransfer.cc | 814 +++++++++++++++++++++++++++++++ src/libstore/datatransfer.hh | 113 +++++ src/libstore/download.cc | 814 ------------------------------- src/libstore/download.hh | 113 ----- src/libstore/http-binary-cache-store.cc | 2 +- src/libstore/s3-binary-cache-store.cc | 2 +- src/nix-channel/nix-channel.cc | 2 +- src/nix-prefetch-url/nix-prefetch-url.cc | 2 +- src/nix/main.cc | 2 +- src/nix/upgrade-nix.cc | 2 +- 15 files changed, 938 insertions(+), 938 deletions(-) create mode 100644 src/libstore/datatransfer.cc create mode 100644 src/libstore/datatransfer.hh delete mode 100644 src/libstore/download.cc delete mode 100644 src/libstore/download.hh (limited to 'src') diff --git a/src/libexpr/common-eval-args.cc b/src/libexpr/common-eval-args.cc index 82bfeac36..26460601d 100644 --- a/src/libexpr/common-eval-args.cc +++ b/src/libexpr/common-eval-args.cc @@ -1,6 +1,6 @@ #include "common-eval-args.hh" #include "shared.hh" -#include "download.hh" +#include "datatransfer.hh" #include "util.hh" #include "eval.hh" #include "fetchers.hh" diff --git a/src/libexpr/eval.cc b/src/libexpr/eval.cc index dac32b6f5..7a20c6fa6 100644 --- a/src/libexpr/eval.cc +++ b/src/libexpr/eval.cc @@ -5,7 +5,7 @@ #include "derivations.hh" #include "globals.hh" #include "eval-inline.hh" -#include "download.hh" +#include "datatransfer.hh" #include "json.hh" #include "function-trace.hh" diff --git a/src/libexpr/parser.y b/src/libexpr/parser.y index 6f25f5cf0..96b6f0ecd 100644 --- a/src/libexpr/parser.y +++ b/src/libexpr/parser.y @@ -544,7 +544,7 @@ formal #include #include "eval.hh" -#include "download.hh" +#include "datatransfer.hh" #include "fetchers.hh" #include "store-api.hh" diff --git a/src/libstore/build.cc b/src/libstore/build.cc index 632982ecd..db01d9510 100644 --- a/src/libstore/build.cc +++ b/src/libstore/build.cc @@ -7,7 +7,7 @@ #include "affinity.hh" #include "builtins.hh" #include "builtins/buildenv.hh" -#include "download.hh" +#include "datatransfer.hh" #include "finally.hh" #include "compression.hh" #include "json.hh" diff --git a/src/libstore/builtins/fetchurl.cc b/src/libstore/builtins/fetchurl.cc index a939f040f..01d21fa98 100644 --- a/src/libstore/builtins/fetchurl.cc +++ b/src/libstore/builtins/fetchurl.cc @@ -1,5 +1,5 @@ #include "builtins.hh" -#include "download.hh" +#include "datatransfer.hh" #include "store-api.hh" #include "archive.hh" #include "compression.hh" diff --git a/src/libstore/datatransfer.cc b/src/libstore/datatransfer.cc new file mode 100644 index 000000000..30ad48577 --- /dev/null +++ b/src/libstore/datatransfer.cc @@ -0,0 +1,814 @@ +#include "datatransfer.hh" +#include "util.hh" +#include "globals.hh" +#include "store-api.hh" +#include "s3.hh" +#include "compression.hh" +#include "finally.hh" + +#ifdef ENABLE_S3 +#include +#endif + +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include + +using namespace std::string_literals; + +namespace nix { + +DownloadSettings downloadSettings; + +static GlobalConfig::Register r1(&downloadSettings); + +std::string resolveUri(const std::string & uri) +{ + if (uri.compare(0, 8, "channel:") == 0) + return "https://nixos.org/channels/" + std::string(uri, 8) + "/nixexprs.tar.xz"; + else + return uri; +} + +struct CurlDownloader : public Downloader +{ + CURLM * curlm = 0; + + std::random_device rd; + std::mt19937 mt19937; + + struct DownloadItem : public std::enable_shared_from_this + { + CurlDownloader & downloader; + DataTransferRequest request; + DownloadResult result; + Activity act; + bool done = false; // whether either the success or failure function has been called + Callback callback; + CURL * req = 0; + bool active = false; // whether the handle has been added to the multi object + std::string status; + + unsigned int attempt = 0; + + /* Don't start this download until the specified time point + has been reached. */ + std::chrono::steady_clock::time_point embargo; + + struct curl_slist * requestHeaders = 0; + + std::string encoding; + + bool acceptRanges = false; + + curl_off_t writtenToSink = 0; + + DownloadItem(CurlDownloader & downloader, + const DataTransferRequest & request, + Callback && callback) + : downloader(downloader) + , request(request) + , act(*logger, lvlTalkative, actDownload, + fmt(request.data ? "uploading '%s'" : "downloading '%s'", request.uri), + {request.uri}, request.parentAct) + , callback(std::move(callback)) + , finalSink([this](const unsigned char * data, size_t len) { + if (this->request.dataCallback) { + long httpStatus = 0; + curl_easy_getinfo(req, CURLINFO_RESPONSE_CODE, &httpStatus); + + /* Only write data to the sink if this is a + successful response. */ + if (httpStatus == 0 || httpStatus == 200 || httpStatus == 201 || httpStatus == 206) { + writtenToSink += len; + this->request.dataCallback((char *) data, len); + } + } else + this->result.data->append((char *) data, len); + }) + { + if (!request.expectedETag.empty()) + requestHeaders = curl_slist_append(requestHeaders, ("If-None-Match: " + request.expectedETag).c_str()); + if (!request.mimeType.empty()) + requestHeaders = curl_slist_append(requestHeaders, ("Content-Type: " + request.mimeType).c_str()); + } + + ~DownloadItem() + { + if (req) { + if (active) + curl_multi_remove_handle(downloader.curlm, req); + curl_easy_cleanup(req); + } + if (requestHeaders) curl_slist_free_all(requestHeaders); + try { + if (!done) + fail(DownloadError(Interrupted, format("download of '%s' was interrupted") % request.uri)); + } catch (...) { + ignoreException(); + } + } + + void failEx(std::exception_ptr ex) + { + assert(!done); + done = true; + callback.rethrow(ex); + } + + template + void fail(const T & e) + { + failEx(std::make_exception_ptr(e)); + } + + LambdaSink finalSink; + std::shared_ptr decompressionSink; + + std::exception_ptr writeException; + + size_t writeCallback(void * contents, size_t size, size_t nmemb) + { + try { + size_t realSize = size * nmemb; + result.bodySize += realSize; + + if (!decompressionSink) + decompressionSink = makeDecompressionSink(encoding, finalSink); + + (*decompressionSink)((unsigned char *) contents, realSize); + + return realSize; + } catch (...) { + writeException = std::current_exception(); + return 0; + } + } + + static size_t writeCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) + { + return ((DownloadItem *) userp)->writeCallback(contents, size, nmemb); + } + + size_t headerCallback(void * contents, size_t size, size_t nmemb) + { + size_t realSize = size * nmemb; + std::string line((char *) contents, realSize); + printMsg(lvlVomit, format("got header for '%s': %s") % request.uri % trim(line)); + if (line.compare(0, 5, "HTTP/") == 0) { // new response starts + result.etag = ""; + auto ss = tokenizeString>(line, " "); + status = ss.size() >= 2 ? ss[1] : ""; + result.data = std::make_shared(); + result.bodySize = 0; + acceptRanges = false; + encoding = ""; + } else { + auto i = line.find(':'); + if (i != string::npos) { + string name = toLower(trim(string(line, 0, i))); + if (name == "etag") { + result.etag = trim(string(line, i + 1)); + /* Hack to work around a GitHub bug: it sends + ETags, but ignores If-None-Match. So if we get + the expected ETag on a 200 response, then shut + down the connection because we already have the + data. */ + if (result.etag == request.expectedETag && status == "200") { + debug(format("shutting down on 200 HTTP response with expected ETag")); + return 0; + } + } else if (name == "content-encoding") + encoding = trim(string(line, i + 1)); + else if (name == "accept-ranges" && toLower(trim(std::string(line, i + 1))) == "bytes") + acceptRanges = true; + } + } + return realSize; + } + + static size_t headerCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) + { + return ((DownloadItem *) userp)->headerCallback(contents, size, nmemb); + } + + int progressCallback(double dltotal, double dlnow) + { + try { + act.progress(dlnow, dltotal); + } catch (nix::Interrupted &) { + assert(_isInterrupted); + } + return _isInterrupted; + } + + static int progressCallbackWrapper(void * userp, double dltotal, double dlnow, double ultotal, double ulnow) + { + return ((DownloadItem *) userp)->progressCallback(dltotal, dlnow); + } + + static int debugCallback(CURL * handle, curl_infotype type, char * data, size_t size, void * userptr) + { + if (type == CURLINFO_TEXT) + vomit("curl: %s", chomp(std::string(data, size))); + return 0; + } + + size_t readOffset = 0; + size_t readCallback(char *buffer, size_t size, size_t nitems) + { + if (readOffset == request.data->length()) + return 0; + auto count = std::min(size * nitems, request.data->length() - readOffset); + assert(count); + memcpy(buffer, request.data->data() + readOffset, count); + readOffset += count; + return count; + } + + static size_t readCallbackWrapper(char *buffer, size_t size, size_t nitems, void * userp) + { + return ((DownloadItem *) userp)->readCallback(buffer, size, nitems); + } + + void init() + { + if (!req) req = curl_easy_init(); + + curl_easy_reset(req); + + if (verbosity >= lvlVomit) { + curl_easy_setopt(req, CURLOPT_VERBOSE, 1); + curl_easy_setopt(req, CURLOPT_DEBUGFUNCTION, DownloadItem::debugCallback); + } + + curl_easy_setopt(req, CURLOPT_URL, request.uri.c_str()); + curl_easy_setopt(req, CURLOPT_FOLLOWLOCATION, 1L); + curl_easy_setopt(req, CURLOPT_MAXREDIRS, 10); + curl_easy_setopt(req, CURLOPT_NOSIGNAL, 1); + curl_easy_setopt(req, CURLOPT_USERAGENT, + ("curl/" LIBCURL_VERSION " Nix/" + nixVersion + + (downloadSettings.userAgentSuffix != "" ? " " + downloadSettings.userAgentSuffix.get() : "")).c_str()); + #if LIBCURL_VERSION_NUM >= 0x072b00 + curl_easy_setopt(req, CURLOPT_PIPEWAIT, 1); + #endif + #if LIBCURL_VERSION_NUM >= 0x072f00 + if (downloadSettings.enableHttp2) + curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS); + else + curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1); + #endif + curl_easy_setopt(req, CURLOPT_WRITEFUNCTION, DownloadItem::writeCallbackWrapper); + curl_easy_setopt(req, CURLOPT_WRITEDATA, this); + curl_easy_setopt(req, CURLOPT_HEADERFUNCTION, DownloadItem::headerCallbackWrapper); + curl_easy_setopt(req, CURLOPT_HEADERDATA, this); + + curl_easy_setopt(req, CURLOPT_PROGRESSFUNCTION, progressCallbackWrapper); + curl_easy_setopt(req, CURLOPT_PROGRESSDATA, this); + curl_easy_setopt(req, CURLOPT_NOPROGRESS, 0); + + curl_easy_setopt(req, CURLOPT_HTTPHEADER, requestHeaders); + + if (request.head) + curl_easy_setopt(req, CURLOPT_NOBODY, 1); + + if (request.data) { + curl_easy_setopt(req, CURLOPT_UPLOAD, 1L); + curl_easy_setopt(req, CURLOPT_READFUNCTION, readCallbackWrapper); + curl_easy_setopt(req, CURLOPT_READDATA, this); + curl_easy_setopt(req, CURLOPT_INFILESIZE_LARGE, (curl_off_t) request.data->length()); + } + + if (request.verifyTLS) { + debug("verify TLS: Nix CA file = '%s'", settings.caFile); + if (settings.caFile != "") + curl_easy_setopt(req, CURLOPT_CAINFO, settings.caFile.c_str()); + } else { + curl_easy_setopt(req, CURLOPT_SSL_VERIFYPEER, 0); + curl_easy_setopt(req, CURLOPT_SSL_VERIFYHOST, 0); + } + + curl_easy_setopt(req, CURLOPT_CONNECTTIMEOUT, downloadSettings.connectTimeout.get()); + + curl_easy_setopt(req, CURLOPT_LOW_SPEED_LIMIT, 1L); + curl_easy_setopt(req, CURLOPT_LOW_SPEED_TIME, downloadSettings.stalledDownloadTimeout.get()); + + /* If no file exist in the specified path, curl continues to work + anyway as if netrc support was disabled. */ + curl_easy_setopt(req, CURLOPT_NETRC_FILE, settings.netrcFile.get().c_str()); + curl_easy_setopt(req, CURLOPT_NETRC, CURL_NETRC_OPTIONAL); + + if (writtenToSink) + curl_easy_setopt(req, CURLOPT_RESUME_FROM_LARGE, writtenToSink); + + result.data = std::make_shared(); + result.bodySize = 0; + } + + void finish(CURLcode code) + { + long httpStatus = 0; + curl_easy_getinfo(req, CURLINFO_RESPONSE_CODE, &httpStatus); + + char * effectiveUriCStr; + curl_easy_getinfo(req, CURLINFO_EFFECTIVE_URL, &effectiveUriCStr); + if (effectiveUriCStr) + result.effectiveUri = effectiveUriCStr; + + debug("finished %s of '%s'; curl status = %d, HTTP status = %d, body = %d bytes", + request.verb(), request.uri, code, httpStatus, result.bodySize); + + if (decompressionSink) { + try { + decompressionSink->finish(); + } catch (...) { + writeException = std::current_exception(); + } + } + + if (code == CURLE_WRITE_ERROR && result.etag == request.expectedETag) { + code = CURLE_OK; + httpStatus = 304; + } + + if (writeException) + failEx(writeException); + + else if (code == CURLE_OK && + (httpStatus == 200 || httpStatus == 201 || httpStatus == 204 || httpStatus == 206 || httpStatus == 304 || httpStatus == 226 /* FTP */ || httpStatus == 0 /* other protocol */)) + { + result.cached = httpStatus == 304; + act.progress(result.bodySize, result.bodySize); + done = true; + callback(std::move(result)); + } + + else { + // We treat most errors as transient, but won't retry when hopeless + Error err = Transient; + + if (httpStatus == 404 || httpStatus == 410 || code == CURLE_FILE_COULDNT_READ_FILE) { + // The file is definitely not there + err = NotFound; + } else if (httpStatus == 401 || httpStatus == 403 || httpStatus == 407) { + // Don't retry on authentication/authorization failures + err = Forbidden; + } else if (httpStatus >= 400 && httpStatus < 500 && httpStatus != 408 && httpStatus != 429) { + // Most 4xx errors are client errors and are probably not worth retrying: + // * 408 means the server timed out waiting for us, so we try again + // * 429 means too many requests, so we retry (with a delay) + err = Misc; + } else if (httpStatus == 501 || httpStatus == 505 || httpStatus == 511) { + // Let's treat most 5xx (server) errors as transient, except for a handful: + // * 501 not implemented + // * 505 http version not supported + // * 511 we're behind a captive portal + err = Misc; + } else { + // Don't bother retrying on certain cURL errors either + switch (code) { + case CURLE_FAILED_INIT: + case CURLE_URL_MALFORMAT: + case CURLE_NOT_BUILT_IN: + case CURLE_REMOTE_ACCESS_DENIED: + case CURLE_FILE_COULDNT_READ_FILE: + case CURLE_FUNCTION_NOT_FOUND: + case CURLE_ABORTED_BY_CALLBACK: + case CURLE_BAD_FUNCTION_ARGUMENT: + case CURLE_INTERFACE_FAILED: + case CURLE_UNKNOWN_OPTION: + case CURLE_SSL_CACERT_BADFILE: + case CURLE_TOO_MANY_REDIRECTS: + case CURLE_WRITE_ERROR: + case CURLE_UNSUPPORTED_PROTOCOL: + err = Misc; + break; + default: // Shut up warnings + break; + } + } + + attempt++; + + auto exc = + code == CURLE_ABORTED_BY_CALLBACK && _isInterrupted + ? DownloadError(Interrupted, fmt("%s of '%s' was interrupted", request.verb(), request.uri)) + : httpStatus != 0 + ? DownloadError(err, + fmt("unable to %s '%s': HTTP error %d", + request.verb(), request.uri, httpStatus) + + (code == CURLE_OK ? "" : fmt(" (curl error: %s)", curl_easy_strerror(code))) + ) + : DownloadError(err, + fmt("unable to %s '%s': %s (%d)", + request.verb(), request.uri, curl_easy_strerror(code), code)); + + /* If this is a transient error, then maybe retry the + download after a while. If we're writing to a + sink, we can only retry if the server supports + ranged requests. */ + if (err == Transient + && attempt < request.tries + && (!this->request.dataCallback + || writtenToSink == 0 + || (acceptRanges && encoding.empty()))) + { + int ms = request.baseRetryTimeMs * std::pow(2.0f, attempt - 1 + std::uniform_real_distribution<>(0.0, 0.5)(downloader.mt19937)); + if (writtenToSink) + warn("%s; retrying from offset %d in %d ms", exc.what(), writtenToSink, ms); + else + warn("%s; retrying in %d ms", exc.what(), ms); + embargo = std::chrono::steady_clock::now() + std::chrono::milliseconds(ms); + downloader.enqueueItem(shared_from_this()); + } + else + fail(exc); + } + } + }; + + struct State + { + struct EmbargoComparator { + bool operator() (const std::shared_ptr & i1, const std::shared_ptr & i2) { + return i1->embargo > i2->embargo; + } + }; + bool quit = false; + std::priority_queue, std::vector>, EmbargoComparator> incoming; + }; + + Sync state_; + + /* We can't use a std::condition_variable to wake up the curl + thread, because it only monitors file descriptors. So use a + pipe instead. */ + Pipe wakeupPipe; + + std::thread workerThread; + + CurlDownloader() + : mt19937(rd()) + { + static std::once_flag globalInit; + std::call_once(globalInit, curl_global_init, CURL_GLOBAL_ALL); + + curlm = curl_multi_init(); + + #if LIBCURL_VERSION_NUM >= 0x072b00 // Multiplex requires >= 7.43.0 + curl_multi_setopt(curlm, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX); + #endif + #if LIBCURL_VERSION_NUM >= 0x071e00 // Max connections requires >= 7.30.0 + curl_multi_setopt(curlm, CURLMOPT_MAX_TOTAL_CONNECTIONS, + downloadSettings.httpConnections.get()); + #endif + + wakeupPipe.create(); + fcntl(wakeupPipe.readSide.get(), F_SETFL, O_NONBLOCK); + + workerThread = std::thread([&]() { workerThreadEntry(); }); + } + + ~CurlDownloader() + { + stopWorkerThread(); + + workerThread.join(); + + if (curlm) curl_multi_cleanup(curlm); + } + + void stopWorkerThread() + { + /* Signal the worker thread to exit. */ + { + auto state(state_.lock()); + state->quit = true; + } + writeFull(wakeupPipe.writeSide.get(), " ", false); + } + + void workerThreadMain() + { + /* Cause this thread to be notified on SIGINT. */ + auto callback = createInterruptCallback([&]() { + stopWorkerThread(); + }); + + std::map> items; + + bool quit = false; + + std::chrono::steady_clock::time_point nextWakeup; + + while (!quit) { + checkInterrupt(); + + /* Let curl do its thing. */ + int running; + CURLMcode mc = curl_multi_perform(curlm, &running); + if (mc != CURLM_OK) + throw nix::Error(format("unexpected error from curl_multi_perform(): %s") % curl_multi_strerror(mc)); + + /* Set the promises of any finished requests. */ + CURLMsg * msg; + int left; + while ((msg = curl_multi_info_read(curlm, &left))) { + if (msg->msg == CURLMSG_DONE) { + auto i = items.find(msg->easy_handle); + assert(i != items.end()); + i->second->finish(msg->data.result); + curl_multi_remove_handle(curlm, i->second->req); + i->second->active = false; + items.erase(i); + } + } + + /* Wait for activity, including wakeup events. */ + int numfds = 0; + struct curl_waitfd extraFDs[1]; + extraFDs[0].fd = wakeupPipe.readSide.get(); + extraFDs[0].events = CURL_WAIT_POLLIN; + extraFDs[0].revents = 0; + long maxSleepTimeMs = items.empty() ? 10000 : 100; + auto sleepTimeMs = + nextWakeup != std::chrono::steady_clock::time_point() + ? std::max(0, (int) std::chrono::duration_cast(nextWakeup - std::chrono::steady_clock::now()).count()) + : maxSleepTimeMs; + vomit("download thread waiting for %d ms", sleepTimeMs); + mc = curl_multi_wait(curlm, extraFDs, 1, sleepTimeMs, &numfds); + if (mc != CURLM_OK) + throw nix::Error(format("unexpected error from curl_multi_wait(): %s") % curl_multi_strerror(mc)); + + nextWakeup = std::chrono::steady_clock::time_point(); + + /* Add new curl requests from the incoming requests queue, + except for requests that are embargoed (waiting for a + retry timeout to expire). */ + if (extraFDs[0].revents & CURL_WAIT_POLLIN) { + char buf[1024]; + auto res = read(extraFDs[0].fd, buf, sizeof(buf)); + if (res == -1 && errno != EINTR) + throw SysError("reading curl wakeup socket"); + } + + std::vector> incoming; + auto now = std::chrono::steady_clock::now(); + + { + auto state(state_.lock()); + while (!state->incoming.empty()) { + auto item = state->incoming.top(); + if (item->embargo <= now) { + incoming.push_back(item); + state->incoming.pop(); + } else { + if (nextWakeup == std::chrono::steady_clock::time_point() + || item->embargo < nextWakeup) + nextWakeup = item->embargo; + break; + } + } + quit = state->quit; + } + + for (auto & item : incoming) { + debug("starting %s of %s", item->request.verb(), item->request.uri); + item->init(); + curl_multi_add_handle(curlm, item->req); + item->active = true; + items[item->req] = item; + } + } + + debug("download thread shutting down"); + } + + void workerThreadEntry() + { + try { + workerThreadMain(); + } catch (nix::Interrupted & e) { + } catch (std::exception & e) { + printError("unexpected error in download thread: %s", e.what()); + } + + { + auto state(state_.lock()); + while (!state->incoming.empty()) state->incoming.pop(); + state->quit = true; + } + } + + void enqueueItem(std::shared_ptr item) + { + if (item->request.data + && !hasPrefix(item->request.uri, "http://") + && !hasPrefix(item->request.uri, "https://")) + throw nix::Error("uploading to '%s' is not supported", item->request.uri); + + { + auto state(state_.lock()); + if (state->quit) + throw nix::Error("cannot enqueue download request because the download thread is shutting down"); + state->incoming.push(item); + } + writeFull(wakeupPipe.writeSide.get(), " "); + } + +#ifdef ENABLE_S3 + std::tuple parseS3Uri(std::string uri) + { + auto [path, params] = splitUriAndParams(uri); + + auto slash = path.find('/', 5); // 5 is the length of "s3://" prefix + if (slash == std::string::npos) + throw nix::Error("bad S3 URI '%s'", path); + + std::string bucketName(path, 5, slash - 5); + std::string key(path, slash + 1); + + return {bucketName, key, params}; + } +#endif + + void enqueueDownload(const DataTransferRequest & request, + Callback callback) override + { + /* Ugly hack to support s3:// URIs. */ + if (hasPrefix(request.uri, "s3://")) { + // FIXME: do this on a worker thread + try { +#ifdef ENABLE_S3 + auto [bucketName, key, params] = parseS3Uri(request.uri); + + std::string profile = get(params, "profile").value_or(""); + std::string region = get(params, "region").value_or(Aws::Region::US_EAST_1); + std::string scheme = get(params, "scheme").value_or(""); + std::string endpoint = get(params, "endpoint").value_or(""); + + S3Helper s3Helper(profile, region, scheme, endpoint); + + // FIXME: implement ETag + auto s3Res = s3Helper.getObject(bucketName, key); + DownloadResult res; + if (!s3Res.data) + throw DownloadError(NotFound, fmt("S3 object '%s' does not exist", request.uri)); + res.data = s3Res.data; + callback(std::move(res)); +#else + throw nix::Error("cannot download '%s' because Nix is not built with S3 support", request.uri); +#endif + } catch (...) { callback.rethrow(); } + return; + } + + enqueueItem(std::make_shared(*this, request, std::move(callback))); + } +}; + +ref getDownloader() +{ + static ref downloader = makeDownloader(); + return downloader; +} + +ref makeDownloader() +{ + return make_ref(); +} + +std::future Downloader::enqueueDownload(const DataTransferRequest & request) +{ + auto promise = std::make_shared>(); + enqueueDownload(request, + {[promise](std::future fut) { + try { + promise->set_value(fut.get()); + } catch (...) { + promise->set_exception(std::current_exception()); + } + }}); + return promise->get_future(); +} + +DownloadResult Downloader::download(const DataTransferRequest & request) +{ + return enqueueDownload(request).get(); +} + +void Downloader::download(DataTransferRequest && request, Sink & sink) +{ + /* Note: we can't call 'sink' via request.dataCallback, because + that would cause the sink to execute on the downloader + thread. If 'sink' is a coroutine, this will fail. Also, if the + sink is expensive (e.g. one that does decompression and writing + to the Nix store), it would stall the download thread too much. + Therefore we use a buffer to communicate data between the + download thread and the calling thread. */ + + struct State { + bool quit = false; + std::exception_ptr exc; + std::string data; + std::condition_variable avail, request; + }; + + auto _state = std::make_shared>(); + + /* In case of an exception, wake up the download thread. FIXME: + abort the download request. */ + Finally finally([&]() { + auto state(_state->lock()); + state->quit = true; + state->request.notify_one(); + }); + + request.dataCallback = [_state](char * buf, size_t len) { + + auto state(_state->lock()); + + if (state->quit) return; + + /* If the buffer is full, then go to sleep until the calling + thread wakes us up (i.e. when it has removed data from the + buffer). We don't wait forever to prevent stalling the + download thread. (Hopefully sleeping will throttle the + sender.) */ + if (state->data.size() > 1024 * 1024) { + debug("download buffer is full; going to sleep"); + state.wait_for(state->request, std::chrono::seconds(10)); + } + + /* Append data to the buffer and wake up the calling + thread. */ + state->data.append(buf, len); + state->avail.notify_one(); + }; + + enqueueDownload(request, + {[_state](std::future fut) { + auto state(_state->lock()); + state->quit = true; + try { + fut.get(); + } catch (...) { + state->exc = std::current_exception(); + } + state->avail.notify_one(); + state->request.notify_one(); + }}); + + while (true) { + checkInterrupt(); + + std::string chunk; + + /* Grab data if available, otherwise wait for the download + thread to wake us up. */ + { + auto state(_state->lock()); + + while (state->data.empty()) { + + if (state->quit) { + if (state->exc) std::rethrow_exception(state->exc); + return; + } + + state.wait(state->avail); + } + + chunk = std::move(state->data); + + state->request.notify_one(); + } + + /* Flush the data to the sink and wake up the download thread + if it's blocked on a full buffer. We don't hold the state + lock while doing this to prevent blocking the download + thread if sink() takes a long time. */ + sink((unsigned char *) chunk.data(), chunk.size()); + } +} + +bool isUri(const string & s) +{ + if (s.compare(0, 8, "channel:") == 0) return true; + size_t pos = s.find("://"); + if (pos == string::npos) return false; + string scheme(s, 0, pos); + return scheme == "http" || scheme == "https" || scheme == "file" || scheme == "channel" || scheme == "git" || scheme == "s3" || scheme == "ssh"; +} + + +} diff --git a/src/libstore/datatransfer.hh b/src/libstore/datatransfer.hh new file mode 100644 index 000000000..89d3fce7f --- /dev/null +++ b/src/libstore/datatransfer.hh @@ -0,0 +1,113 @@ +#pragma once + +#include "types.hh" +#include "hash.hh" +#include "config.hh" + +#include +#include + +namespace nix { + +struct DownloadSettings : Config +{ + Setting enableHttp2{this, true, "http2", + "Whether to enable HTTP/2 support."}; + + Setting userAgentSuffix{this, "", "user-agent-suffix", + "String appended to the user agent in HTTP requests."}; + + Setting httpConnections{this, 25, "http-connections", + "Number of parallel HTTP connections.", + {"binary-caches-parallel-connections"}}; + + Setting connectTimeout{this, 0, "connect-timeout", + "Timeout for connecting to servers during downloads. 0 means use curl's builtin default."}; + + Setting stalledDownloadTimeout{this, 300, "stalled-download-timeout", + "Timeout (in seconds) for receiving data from servers during download. Nix cancels idle downloads after this timeout's duration."}; + + Setting tries{this, 5, "download-attempts", + "How often Nix will attempt to download a file before giving up."}; +}; + +extern DownloadSettings downloadSettings; + +struct DataTransferRequest +{ + std::string uri; + std::string expectedETag; + bool verifyTLS = true; + bool head = false; + size_t tries = downloadSettings.tries; + unsigned int baseRetryTimeMs = 250; + ActivityId parentAct; + bool decompress = true; + std::shared_ptr data; + std::string mimeType; + std::function dataCallback; + + DataTransferRequest(const std::string & uri) + : uri(uri), parentAct(getCurActivity()) { } + + std::string verb() + { + return data ? "upload" : "download"; + } +}; + +struct DownloadResult +{ + bool cached = false; + std::string etag; + std::string effectiveUri; + std::shared_ptr data; + uint64_t bodySize = 0; +}; + +class Store; + +struct Downloader +{ + virtual ~Downloader() { } + + /* Enqueue a download request, returning a future to the result of + the download. The future may throw a DownloadError + exception. */ + virtual void enqueueDownload(const DataTransferRequest & request, + Callback callback) = 0; + + std::future enqueueDownload(const DataTransferRequest & request); + + /* Synchronously download a file. */ + DownloadResult download(const DataTransferRequest & request); + + /* Download a file, writing its data to a sink. The sink will be + invoked on the thread of the caller. */ + void download(DataTransferRequest && request, Sink & sink); + + enum Error { NotFound, Forbidden, Misc, Transient, Interrupted }; +}; + +/* Return a shared Downloader object. Using this object is preferred + because it enables connection reuse and HTTP/2 multiplexing. */ +ref getDownloader(); + +/* Return a new Downloader object. */ +ref makeDownloader(); + +class DownloadError : public Error +{ +public: + Downloader::Error error; + DownloadError(Downloader::Error error, const FormatOrString & fs) + : Error(fs), error(error) + { } +}; + +bool isUri(const string & s); + +/* Resolve deprecated 'channel:' URLs. */ +std::string resolveUri(const std::string & uri); + +} diff --git a/src/libstore/download.cc b/src/libstore/download.cc deleted file mode 100644 index 77f3c527a..000000000 --- a/src/libstore/download.cc +++ /dev/null @@ -1,814 +0,0 @@ -#include "download.hh" -#include "util.hh" -#include "globals.hh" -#include "store-api.hh" -#include "s3.hh" -#include "compression.hh" -#include "finally.hh" - -#ifdef ENABLE_S3 -#include -#endif - -#include -#include - -#include - -#include -#include -#include -#include -#include -#include -#include - -using namespace std::string_literals; - -namespace nix { - -DownloadSettings downloadSettings; - -static GlobalConfig::Register r1(&downloadSettings); - -std::string resolveUri(const std::string & uri) -{ - if (uri.compare(0, 8, "channel:") == 0) - return "https://nixos.org/channels/" + std::string(uri, 8) + "/nixexprs.tar.xz"; - else - return uri; -} - -struct CurlDownloader : public Downloader -{ - CURLM * curlm = 0; - - std::random_device rd; - std::mt19937 mt19937; - - struct DownloadItem : public std::enable_shared_from_this - { - CurlDownloader & downloader; - DataTransferRequest request; - DownloadResult result; - Activity act; - bool done = false; // whether either the success or failure function has been called - Callback callback; - CURL * req = 0; - bool active = false; // whether the handle has been added to the multi object - std::string status; - - unsigned int attempt = 0; - - /* Don't start this download until the specified time point - has been reached. */ - std::chrono::steady_clock::time_point embargo; - - struct curl_slist * requestHeaders = 0; - - std::string encoding; - - bool acceptRanges = false; - - curl_off_t writtenToSink = 0; - - DownloadItem(CurlDownloader & downloader, - const DataTransferRequest & request, - Callback && callback) - : downloader(downloader) - , request(request) - , act(*logger, lvlTalkative, actDownload, - fmt(request.data ? "uploading '%s'" : "downloading '%s'", request.uri), - {request.uri}, request.parentAct) - , callback(std::move(callback)) - , finalSink([this](const unsigned char * data, size_t len) { - if (this->request.dataCallback) { - long httpStatus = 0; - curl_easy_getinfo(req, CURLINFO_RESPONSE_CODE, &httpStatus); - - /* Only write data to the sink if this is a - successful response. */ - if (httpStatus == 0 || httpStatus == 200 || httpStatus == 201 || httpStatus == 206) { - writtenToSink += len; - this->request.dataCallback((char *) data, len); - } - } else - this->result.data->append((char *) data, len); - }) - { - if (!request.expectedETag.empty()) - requestHeaders = curl_slist_append(requestHeaders, ("If-None-Match: " + request.expectedETag).c_str()); - if (!request.mimeType.empty()) - requestHeaders = curl_slist_append(requestHeaders, ("Content-Type: " + request.mimeType).c_str()); - } - - ~DownloadItem() - { - if (req) { - if (active) - curl_multi_remove_handle(downloader.curlm, req); - curl_easy_cleanup(req); - } - if (requestHeaders) curl_slist_free_all(requestHeaders); - try { - if (!done) - fail(DownloadError(Interrupted, format("download of '%s' was interrupted") % request.uri)); - } catch (...) { - ignoreException(); - } - } - - void failEx(std::exception_ptr ex) - { - assert(!done); - done = true; - callback.rethrow(ex); - } - - template - void fail(const T & e) - { - failEx(std::make_exception_ptr(e)); - } - - LambdaSink finalSink; - std::shared_ptr decompressionSink; - - std::exception_ptr writeException; - - size_t writeCallback(void * contents, size_t size, size_t nmemb) - { - try { - size_t realSize = size * nmemb; - result.bodySize += realSize; - - if (!decompressionSink) - decompressionSink = makeDecompressionSink(encoding, finalSink); - - (*decompressionSink)((unsigned char *) contents, realSize); - - return realSize; - } catch (...) { - writeException = std::current_exception(); - return 0; - } - } - - static size_t writeCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) - { - return ((DownloadItem *) userp)->writeCallback(contents, size, nmemb); - } - - size_t headerCallback(void * contents, size_t size, size_t nmemb) - { - size_t realSize = size * nmemb; - std::string line((char *) contents, realSize); - printMsg(lvlVomit, format("got header for '%s': %s") % request.uri % trim(line)); - if (line.compare(0, 5, "HTTP/") == 0) { // new response starts - result.etag = ""; - auto ss = tokenizeString>(line, " "); - status = ss.size() >= 2 ? ss[1] : ""; - result.data = std::make_shared(); - result.bodySize = 0; - acceptRanges = false; - encoding = ""; - } else { - auto i = line.find(':'); - if (i != string::npos) { - string name = toLower(trim(string(line, 0, i))); - if (name == "etag") { - result.etag = trim(string(line, i + 1)); - /* Hack to work around a GitHub bug: it sends - ETags, but ignores If-None-Match. So if we get - the expected ETag on a 200 response, then shut - down the connection because we already have the - data. */ - if (result.etag == request.expectedETag && status == "200") { - debug(format("shutting down on 200 HTTP response with expected ETag")); - return 0; - } - } else if (name == "content-encoding") - encoding = trim(string(line, i + 1)); - else if (name == "accept-ranges" && toLower(trim(std::string(line, i + 1))) == "bytes") - acceptRanges = true; - } - } - return realSize; - } - - static size_t headerCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) - { - return ((DownloadItem *) userp)->headerCallback(contents, size, nmemb); - } - - int progressCallback(double dltotal, double dlnow) - { - try { - act.progress(dlnow, dltotal); - } catch (nix::Interrupted &) { - assert(_isInterrupted); - } - return _isInterrupted; - } - - static int progressCallbackWrapper(void * userp, double dltotal, double dlnow, double ultotal, double ulnow) - { - return ((DownloadItem *) userp)->progressCallback(dltotal, dlnow); - } - - static int debugCallback(CURL * handle, curl_infotype type, char * data, size_t size, void * userptr) - { - if (type == CURLINFO_TEXT) - vomit("curl: %s", chomp(std::string(data, size))); - return 0; - } - - size_t readOffset = 0; - size_t readCallback(char *buffer, size_t size, size_t nitems) - { - if (readOffset == request.data->length()) - return 0; - auto count = std::min(size * nitems, request.data->length() - readOffset); - assert(count); - memcpy(buffer, request.data->data() + readOffset, count); - readOffset += count; - return count; - } - - static size_t readCallbackWrapper(char *buffer, size_t size, size_t nitems, void * userp) - { - return ((DownloadItem *) userp)->readCallback(buffer, size, nitems); - } - - void init() - { - if (!req) req = curl_easy_init(); - - curl_easy_reset(req); - - if (verbosity >= lvlVomit) { - curl_easy_setopt(req, CURLOPT_VERBOSE, 1); - curl_easy_setopt(req, CURLOPT_DEBUGFUNCTION, DownloadItem::debugCallback); - } - - curl_easy_setopt(req, CURLOPT_URL, request.uri.c_str()); - curl_easy_setopt(req, CURLOPT_FOLLOWLOCATION, 1L); - curl_easy_setopt(req, CURLOPT_MAXREDIRS, 10); - curl_easy_setopt(req, CURLOPT_NOSIGNAL, 1); - curl_easy_setopt(req, CURLOPT_USERAGENT, - ("curl/" LIBCURL_VERSION " Nix/" + nixVersion + - (downloadSettings.userAgentSuffix != "" ? " " + downloadSettings.userAgentSuffix.get() : "")).c_str()); - #if LIBCURL_VERSION_NUM >= 0x072b00 - curl_easy_setopt(req, CURLOPT_PIPEWAIT, 1); - #endif - #if LIBCURL_VERSION_NUM >= 0x072f00 - if (downloadSettings.enableHttp2) - curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS); - else - curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1); - #endif - curl_easy_setopt(req, CURLOPT_WRITEFUNCTION, DownloadItem::writeCallbackWrapper); - curl_easy_setopt(req, CURLOPT_WRITEDATA, this); - curl_easy_setopt(req, CURLOPT_HEADERFUNCTION, DownloadItem::headerCallbackWrapper); - curl_easy_setopt(req, CURLOPT_HEADERDATA, this); - - curl_easy_setopt(req, CURLOPT_PROGRESSFUNCTION, progressCallbackWrapper); - curl_easy_setopt(req, CURLOPT_PROGRESSDATA, this); - curl_easy_setopt(req, CURLOPT_NOPROGRESS, 0); - - curl_easy_setopt(req, CURLOPT_HTTPHEADER, requestHeaders); - - if (request.head) - curl_easy_setopt(req, CURLOPT_NOBODY, 1); - - if (request.data) { - curl_easy_setopt(req, CURLOPT_UPLOAD, 1L); - curl_easy_setopt(req, CURLOPT_READFUNCTION, readCallbackWrapper); - curl_easy_setopt(req, CURLOPT_READDATA, this); - curl_easy_setopt(req, CURLOPT_INFILESIZE_LARGE, (curl_off_t) request.data->length()); - } - - if (request.verifyTLS) { - debug("verify TLS: Nix CA file = '%s'", settings.caFile); - if (settings.caFile != "") - curl_easy_setopt(req, CURLOPT_CAINFO, settings.caFile.c_str()); - } else { - curl_easy_setopt(req, CURLOPT_SSL_VERIFYPEER, 0); - curl_easy_setopt(req, CURLOPT_SSL_VERIFYHOST, 0); - } - - curl_easy_setopt(req, CURLOPT_CONNECTTIMEOUT, downloadSettings.connectTimeout.get()); - - curl_easy_setopt(req, CURLOPT_LOW_SPEED_LIMIT, 1L); - curl_easy_setopt(req, CURLOPT_LOW_SPEED_TIME, downloadSettings.stalledDownloadTimeout.get()); - - /* If no file exist in the specified path, curl continues to work - anyway as if netrc support was disabled. */ - curl_easy_setopt(req, CURLOPT_NETRC_FILE, settings.netrcFile.get().c_str()); - curl_easy_setopt(req, CURLOPT_NETRC, CURL_NETRC_OPTIONAL); - - if (writtenToSink) - curl_easy_setopt(req, CURLOPT_RESUME_FROM_LARGE, writtenToSink); - - result.data = std::make_shared(); - result.bodySize = 0; - } - - void finish(CURLcode code) - { - long httpStatus = 0; - curl_easy_getinfo(req, CURLINFO_RESPONSE_CODE, &httpStatus); - - char * effectiveUriCStr; - curl_easy_getinfo(req, CURLINFO_EFFECTIVE_URL, &effectiveUriCStr); - if (effectiveUriCStr) - result.effectiveUri = effectiveUriCStr; - - debug("finished %s of '%s'; curl status = %d, HTTP status = %d, body = %d bytes", - request.verb(), request.uri, code, httpStatus, result.bodySize); - - if (decompressionSink) { - try { - decompressionSink->finish(); - } catch (...) { - writeException = std::current_exception(); - } - } - - if (code == CURLE_WRITE_ERROR && result.etag == request.expectedETag) { - code = CURLE_OK; - httpStatus = 304; - } - - if (writeException) - failEx(writeException); - - else if (code == CURLE_OK && - (httpStatus == 200 || httpStatus == 201 || httpStatus == 204 || httpStatus == 206 || httpStatus == 304 || httpStatus == 226 /* FTP */ || httpStatus == 0 /* other protocol */)) - { - result.cached = httpStatus == 304; - act.progress(result.bodySize, result.bodySize); - done = true; - callback(std::move(result)); - } - - else { - // We treat most errors as transient, but won't retry when hopeless - Error err = Transient; - - if (httpStatus == 404 || httpStatus == 410 || code == CURLE_FILE_COULDNT_READ_FILE) { - // The file is definitely not there - err = NotFound; - } else if (httpStatus == 401 || httpStatus == 403 || httpStatus == 407) { - // Don't retry on authentication/authorization failures - err = Forbidden; - } else if (httpStatus >= 400 && httpStatus < 500 && httpStatus != 408 && httpStatus != 429) { - // Most 4xx errors are client errors and are probably not worth retrying: - // * 408 means the server timed out waiting for us, so we try again - // * 429 means too many requests, so we retry (with a delay) - err = Misc; - } else if (httpStatus == 501 || httpStatus == 505 || httpStatus == 511) { - // Let's treat most 5xx (server) errors as transient, except for a handful: - // * 501 not implemented - // * 505 http version not supported - // * 511 we're behind a captive portal - err = Misc; - } else { - // Don't bother retrying on certain cURL errors either - switch (code) { - case CURLE_FAILED_INIT: - case CURLE_URL_MALFORMAT: - case CURLE_NOT_BUILT_IN: - case CURLE_REMOTE_ACCESS_DENIED: - case CURLE_FILE_COULDNT_READ_FILE: - case CURLE_FUNCTION_NOT_FOUND: - case CURLE_ABORTED_BY_CALLBACK: - case CURLE_BAD_FUNCTION_ARGUMENT: - case CURLE_INTERFACE_FAILED: - case CURLE_UNKNOWN_OPTION: - case CURLE_SSL_CACERT_BADFILE: - case CURLE_TOO_MANY_REDIRECTS: - case CURLE_WRITE_ERROR: - case CURLE_UNSUPPORTED_PROTOCOL: - err = Misc; - break; - default: // Shut up warnings - break; - } - } - - attempt++; - - auto exc = - code == CURLE_ABORTED_BY_CALLBACK && _isInterrupted - ? DownloadError(Interrupted, fmt("%s of '%s' was interrupted", request.verb(), request.uri)) - : httpStatus != 0 - ? DownloadError(err, - fmt("unable to %s '%s': HTTP error %d", - request.verb(), request.uri, httpStatus) - + (code == CURLE_OK ? "" : fmt(" (curl error: %s)", curl_easy_strerror(code))) - ) - : DownloadError(err, - fmt("unable to %s '%s': %s (%d)", - request.verb(), request.uri, curl_easy_strerror(code), code)); - - /* If this is a transient error, then maybe retry the - download after a while. If we're writing to a - sink, we can only retry if the server supports - ranged requests. */ - if (err == Transient - && attempt < request.tries - && (!this->request.dataCallback - || writtenToSink == 0 - || (acceptRanges && encoding.empty()))) - { - int ms = request.baseRetryTimeMs * std::pow(2.0f, attempt - 1 + std::uniform_real_distribution<>(0.0, 0.5)(downloader.mt19937)); - if (writtenToSink) - warn("%s; retrying from offset %d in %d ms", exc.what(), writtenToSink, ms); - else - warn("%s; retrying in %d ms", exc.what(), ms); - embargo = std::chrono::steady_clock::now() + std::chrono::milliseconds(ms); - downloader.enqueueItem(shared_from_this()); - } - else - fail(exc); - } - } - }; - - struct State - { - struct EmbargoComparator { - bool operator() (const std::shared_ptr & i1, const std::shared_ptr & i2) { - return i1->embargo > i2->embargo; - } - }; - bool quit = false; - std::priority_queue, std::vector>, EmbargoComparator> incoming; - }; - - Sync state_; - - /* We can't use a std::condition_variable to wake up the curl - thread, because it only monitors file descriptors. So use a - pipe instead. */ - Pipe wakeupPipe; - - std::thread workerThread; - - CurlDownloader() - : mt19937(rd()) - { - static std::once_flag globalInit; - std::call_once(globalInit, curl_global_init, CURL_GLOBAL_ALL); - - curlm = curl_multi_init(); - - #if LIBCURL_VERSION_NUM >= 0x072b00 // Multiplex requires >= 7.43.0 - curl_multi_setopt(curlm, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX); - #endif - #if LIBCURL_VERSION_NUM >= 0x071e00 // Max connections requires >= 7.30.0 - curl_multi_setopt(curlm, CURLMOPT_MAX_TOTAL_CONNECTIONS, - downloadSettings.httpConnections.get()); - #endif - - wakeupPipe.create(); - fcntl(wakeupPipe.readSide.get(), F_SETFL, O_NONBLOCK); - - workerThread = std::thread([&]() { workerThreadEntry(); }); - } - - ~CurlDownloader() - { - stopWorkerThread(); - - workerThread.join(); - - if (curlm) curl_multi_cleanup(curlm); - } - - void stopWorkerThread() - { - /* Signal the worker thread to exit. */ - { - auto state(state_.lock()); - state->quit = true; - } - writeFull(wakeupPipe.writeSide.get(), " ", false); - } - - void workerThreadMain() - { - /* Cause this thread to be notified on SIGINT. */ - auto callback = createInterruptCallback([&]() { - stopWorkerThread(); - }); - - std::map> items; - - bool quit = false; - - std::chrono::steady_clock::time_point nextWakeup; - - while (!quit) { - checkInterrupt(); - - /* Let curl do its thing. */ - int running; - CURLMcode mc = curl_multi_perform(curlm, &running); - if (mc != CURLM_OK) - throw nix::Error(format("unexpected error from curl_multi_perform(): %s") % curl_multi_strerror(mc)); - - /* Set the promises of any finished requests. */ - CURLMsg * msg; - int left; - while ((msg = curl_multi_info_read(curlm, &left))) { - if (msg->msg == CURLMSG_DONE) { - auto i = items.find(msg->easy_handle); - assert(i != items.end()); - i->second->finish(msg->data.result); - curl_multi_remove_handle(curlm, i->second->req); - i->second->active = false; - items.erase(i); - } - } - - /* Wait for activity, including wakeup events. */ - int numfds = 0; - struct curl_waitfd extraFDs[1]; - extraFDs[0].fd = wakeupPipe.readSide.get(); - extraFDs[0].events = CURL_WAIT_POLLIN; - extraFDs[0].revents = 0; - long maxSleepTimeMs = items.empty() ? 10000 : 100; - auto sleepTimeMs = - nextWakeup != std::chrono::steady_clock::time_point() - ? std::max(0, (int) std::chrono::duration_cast(nextWakeup - std::chrono::steady_clock::now()).count()) - : maxSleepTimeMs; - vomit("download thread waiting for %d ms", sleepTimeMs); - mc = curl_multi_wait(curlm, extraFDs, 1, sleepTimeMs, &numfds); - if (mc != CURLM_OK) - throw nix::Error(format("unexpected error from curl_multi_wait(): %s") % curl_multi_strerror(mc)); - - nextWakeup = std::chrono::steady_clock::time_point(); - - /* Add new curl requests from the incoming requests queue, - except for requests that are embargoed (waiting for a - retry timeout to expire). */ - if (extraFDs[0].revents & CURL_WAIT_POLLIN) { - char buf[1024]; - auto res = read(extraFDs[0].fd, buf, sizeof(buf)); - if (res == -1 && errno != EINTR) - throw SysError("reading curl wakeup socket"); - } - - std::vector> incoming; - auto now = std::chrono::steady_clock::now(); - - { - auto state(state_.lock()); - while (!state->incoming.empty()) { - auto item = state->incoming.top(); - if (item->embargo <= now) { - incoming.push_back(item); - state->incoming.pop(); - } else { - if (nextWakeup == std::chrono::steady_clock::time_point() - || item->embargo < nextWakeup) - nextWakeup = item->embargo; - break; - } - } - quit = state->quit; - } - - for (auto & item : incoming) { - debug("starting %s of %s", item->request.verb(), item->request.uri); - item->init(); - curl_multi_add_handle(curlm, item->req); - item->active = true; - items[item->req] = item; - } - } - - debug("download thread shutting down"); - } - - void workerThreadEntry() - { - try { - workerThreadMain(); - } catch (nix::Interrupted & e) { - } catch (std::exception & e) { - printError("unexpected error in download thread: %s", e.what()); - } - - { - auto state(state_.lock()); - while (!state->incoming.empty()) state->incoming.pop(); - state->quit = true; - } - } - - void enqueueItem(std::shared_ptr item) - { - if (item->request.data - && !hasPrefix(item->request.uri, "http://") - && !hasPrefix(item->request.uri, "https://")) - throw nix::Error("uploading to '%s' is not supported", item->request.uri); - - { - auto state(state_.lock()); - if (state->quit) - throw nix::Error("cannot enqueue download request because the download thread is shutting down"); - state->incoming.push(item); - } - writeFull(wakeupPipe.writeSide.get(), " "); - } - -#ifdef ENABLE_S3 - std::tuple parseS3Uri(std::string uri) - { - auto [path, params] = splitUriAndParams(uri); - - auto slash = path.find('/', 5); // 5 is the length of "s3://" prefix - if (slash == std::string::npos) - throw nix::Error("bad S3 URI '%s'", path); - - std::string bucketName(path, 5, slash - 5); - std::string key(path, slash + 1); - - return {bucketName, key, params}; - } -#endif - - void enqueueDownload(const DataTransferRequest & request, - Callback callback) override - { - /* Ugly hack to support s3:// URIs. */ - if (hasPrefix(request.uri, "s3://")) { - // FIXME: do this on a worker thread - try { -#ifdef ENABLE_S3 - auto [bucketName, key, params] = parseS3Uri(request.uri); - - std::string profile = get(params, "profile").value_or(""); - std::string region = get(params, "region").value_or(Aws::Region::US_EAST_1); - std::string scheme = get(params, "scheme").value_or(""); - std::string endpoint = get(params, "endpoint").value_or(""); - - S3Helper s3Helper(profile, region, scheme, endpoint); - - // FIXME: implement ETag - auto s3Res = s3Helper.getObject(bucketName, key); - DownloadResult res; - if (!s3Res.data) - throw DownloadError(NotFound, fmt("S3 object '%s' does not exist", request.uri)); - res.data = s3Res.data; - callback(std::move(res)); -#else - throw nix::Error("cannot download '%s' because Nix is not built with S3 support", request.uri); -#endif - } catch (...) { callback.rethrow(); } - return; - } - - enqueueItem(std::make_shared(*this, request, std::move(callback))); - } -}; - -ref getDownloader() -{ - static ref downloader = makeDownloader(); - return downloader; -} - -ref makeDownloader() -{ - return make_ref(); -} - -std::future Downloader::enqueueDownload(const DataTransferRequest & request) -{ - auto promise = std::make_shared>(); - enqueueDownload(request, - {[promise](std::future fut) { - try { - promise->set_value(fut.get()); - } catch (...) { - promise->set_exception(std::current_exception()); - } - }}); - return promise->get_future(); -} - -DownloadResult Downloader::download(const DataTransferRequest & request) -{ - return enqueueDownload(request).get(); -} - -void Downloader::download(DataTransferRequest && request, Sink & sink) -{ - /* Note: we can't call 'sink' via request.dataCallback, because - that would cause the sink to execute on the downloader - thread. If 'sink' is a coroutine, this will fail. Also, if the - sink is expensive (e.g. one that does decompression and writing - to the Nix store), it would stall the download thread too much. - Therefore we use a buffer to communicate data between the - download thread and the calling thread. */ - - struct State { - bool quit = false; - std::exception_ptr exc; - std::string data; - std::condition_variable avail, request; - }; - - auto _state = std::make_shared>(); - - /* In case of an exception, wake up the download thread. FIXME: - abort the download request. */ - Finally finally([&]() { - auto state(_state->lock()); - state->quit = true; - state->request.notify_one(); - }); - - request.dataCallback = [_state](char * buf, size_t len) { - - auto state(_state->lock()); - - if (state->quit) return; - - /* If the buffer is full, then go to sleep until the calling - thread wakes us up (i.e. when it has removed data from the - buffer). We don't wait forever to prevent stalling the - download thread. (Hopefully sleeping will throttle the - sender.) */ - if (state->data.size() > 1024 * 1024) { - debug("download buffer is full; going to sleep"); - state.wait_for(state->request, std::chrono::seconds(10)); - } - - /* Append data to the buffer and wake up the calling - thread. */ - state->data.append(buf, len); - state->avail.notify_one(); - }; - - enqueueDownload(request, - {[_state](std::future fut) { - auto state(_state->lock()); - state->quit = true; - try { - fut.get(); - } catch (...) { - state->exc = std::current_exception(); - } - state->avail.notify_one(); - state->request.notify_one(); - }}); - - while (true) { - checkInterrupt(); - - std::string chunk; - - /* Grab data if available, otherwise wait for the download - thread to wake us up. */ - { - auto state(_state->lock()); - - while (state->data.empty()) { - - if (state->quit) { - if (state->exc) std::rethrow_exception(state->exc); - return; - } - - state.wait(state->avail); - } - - chunk = std::move(state->data); - - state->request.notify_one(); - } - - /* Flush the data to the sink and wake up the download thread - if it's blocked on a full buffer. We don't hold the state - lock while doing this to prevent blocking the download - thread if sink() takes a long time. */ - sink((unsigned char *) chunk.data(), chunk.size()); - } -} - -bool isUri(const string & s) -{ - if (s.compare(0, 8, "channel:") == 0) return true; - size_t pos = s.find("://"); - if (pos == string::npos) return false; - string scheme(s, 0, pos); - return scheme == "http" || scheme == "https" || scheme == "file" || scheme == "channel" || scheme == "git" || scheme == "s3" || scheme == "ssh"; -} - - -} diff --git a/src/libstore/download.hh b/src/libstore/download.hh deleted file mode 100644 index 89d3fce7f..000000000 --- a/src/libstore/download.hh +++ /dev/null @@ -1,113 +0,0 @@ -#pragma once - -#include "types.hh" -#include "hash.hh" -#include "config.hh" - -#include -#include - -namespace nix { - -struct DownloadSettings : Config -{ - Setting enableHttp2{this, true, "http2", - "Whether to enable HTTP/2 support."}; - - Setting userAgentSuffix{this, "", "user-agent-suffix", - "String appended to the user agent in HTTP requests."}; - - Setting httpConnections{this, 25, "http-connections", - "Number of parallel HTTP connections.", - {"binary-caches-parallel-connections"}}; - - Setting connectTimeout{this, 0, "connect-timeout", - "Timeout for connecting to servers during downloads. 0 means use curl's builtin default."}; - - Setting stalledDownloadTimeout{this, 300, "stalled-download-timeout", - "Timeout (in seconds) for receiving data from servers during download. Nix cancels idle downloads after this timeout's duration."}; - - Setting tries{this, 5, "download-attempts", - "How often Nix will attempt to download a file before giving up."}; -}; - -extern DownloadSettings downloadSettings; - -struct DataTransferRequest -{ - std::string uri; - std::string expectedETag; - bool verifyTLS = true; - bool head = false; - size_t tries = downloadSettings.tries; - unsigned int baseRetryTimeMs = 250; - ActivityId parentAct; - bool decompress = true; - std::shared_ptr data; - std::string mimeType; - std::function dataCallback; - - DataTransferRequest(const std::string & uri) - : uri(uri), parentAct(getCurActivity()) { } - - std::string verb() - { - return data ? "upload" : "download"; - } -}; - -struct DownloadResult -{ - bool cached = false; - std::string etag; - std::string effectiveUri; - std::shared_ptr data; - uint64_t bodySize = 0; -}; - -class Store; - -struct Downloader -{ - virtual ~Downloader() { } - - /* Enqueue a download request, returning a future to the result of - the download. The future may throw a DownloadError - exception. */ - virtual void enqueueDownload(const DataTransferRequest & request, - Callback callback) = 0; - - std::future enqueueDownload(const DataTransferRequest & request); - - /* Synchronously download a file. */ - DownloadResult download(const DataTransferRequest & request); - - /* Download a file, writing its data to a sink. The sink will be - invoked on the thread of the caller. */ - void download(DataTransferRequest && request, Sink & sink); - - enum Error { NotFound, Forbidden, Misc, Transient, Interrupted }; -}; - -/* Return a shared Downloader object. Using this object is preferred - because it enables connection reuse and HTTP/2 multiplexing. */ -ref getDownloader(); - -/* Return a new Downloader object. */ -ref makeDownloader(); - -class DownloadError : public Error -{ -public: - Downloader::Error error; - DownloadError(Downloader::Error error, const FormatOrString & fs) - : Error(fs), error(error) - { } -}; - -bool isUri(const string & s); - -/* Resolve deprecated 'channel:' URLs. */ -std::string resolveUri(const std::string & uri); - -} diff --git a/src/libstore/http-binary-cache-store.cc b/src/libstore/http-binary-cache-store.cc index e152e7524..1fccee518 100644 --- a/src/libstore/http-binary-cache-store.cc +++ b/src/libstore/http-binary-cache-store.cc @@ -1,5 +1,5 @@ #include "binary-cache-store.hh" -#include "download.hh" +#include "datatransfer.hh" #include "globals.hh" #include "nar-info-disk-cache.hh" diff --git a/src/libstore/s3-binary-cache-store.cc b/src/libstore/s3-binary-cache-store.cc index f2e4b63e0..ed3c690b6 100644 --- a/src/libstore/s3-binary-cache-store.cc +++ b/src/libstore/s3-binary-cache-store.cc @@ -6,7 +6,7 @@ #include "nar-info-disk-cache.hh" #include "globals.hh" #include "compression.hh" -#include "download.hh" +#include "datatransfer.hh" #include "istringstream_nocopy.hh" #include diff --git a/src/nix-channel/nix-channel.cc b/src/nix-channel/nix-channel.cc index 2a9defb4e..d0719194d 100755 --- a/src/nix-channel/nix-channel.cc +++ b/src/nix-channel/nix-channel.cc @@ -1,6 +1,6 @@ #include "shared.hh" #include "globals.hh" -#include "download.hh" +#include "datatransfer.hh" #include "store-api.hh" #include "../nix/legacy.hh" #include "fetchers.hh" diff --git a/src/nix-prefetch-url/nix-prefetch-url.cc b/src/nix-prefetch-url/nix-prefetch-url.cc index bfd65da98..58a180e72 100644 --- a/src/nix-prefetch-url/nix-prefetch-url.cc +++ b/src/nix-prefetch-url/nix-prefetch-url.cc @@ -1,6 +1,6 @@ #include "hash.hh" #include "shared.hh" -#include "download.hh" +#include "datatransfer.hh" #include "store-api.hh" #include "eval.hh" #include "eval-inline.hh" diff --git a/src/nix/main.cc b/src/nix/main.cc index 3b5f5516f..ad4102201 100644 --- a/src/nix/main.cc +++ b/src/nix/main.cc @@ -8,7 +8,7 @@ #include "shared.hh" #include "store-api.hh" #include "progress-bar.hh" -#include "download.hh" +#include "datatransfer.hh" #include "finally.hh" #include diff --git a/src/nix/upgrade-nix.cc b/src/nix/upgrade-nix.cc index aeaf7b09c..414689ec5 100644 --- a/src/nix/upgrade-nix.cc +++ b/src/nix/upgrade-nix.cc @@ -1,7 +1,7 @@ #include "command.hh" #include "common-args.hh" #include "store-api.hh" -#include "download.hh" +#include "datatransfer.hh" #include "eval.hh" #include "attr-path.hh" #include "names.hh" -- cgit v1.2.3 From e5cc53beec62a36ebdc219e26b87f501ddb84a4f Mon Sep 17 00:00:00 2001 From: Nikola Knezevic Date: Mon, 6 Apr 2020 15:14:49 +0200 Subject: DownloadSettings -> DataTransferSettings --- src/libstore/datatransfer.cc | 14 +++++++------- src/libstore/datatransfer.hh | 6 +++--- src/nix/main.cc | 8 ++++---- 3 files changed, 14 insertions(+), 14 deletions(-) (limited to 'src') diff --git a/src/libstore/datatransfer.cc b/src/libstore/datatransfer.cc index 30ad48577..62f546cfa 100644 --- a/src/libstore/datatransfer.cc +++ b/src/libstore/datatransfer.cc @@ -27,9 +27,9 @@ using namespace std::string_literals; namespace nix { -DownloadSettings downloadSettings; +DataTransferSettings dataTransferSettings; -static GlobalConfig::Register r1(&downloadSettings); +static GlobalConfig::Register r1(&dataTransferSettings); std::string resolveUri(const std::string & uri) { @@ -257,12 +257,12 @@ struct CurlDownloader : public Downloader curl_easy_setopt(req, CURLOPT_NOSIGNAL, 1); curl_easy_setopt(req, CURLOPT_USERAGENT, ("curl/" LIBCURL_VERSION " Nix/" + nixVersion + - (downloadSettings.userAgentSuffix != "" ? " " + downloadSettings.userAgentSuffix.get() : "")).c_str()); + (dataTransferSettings.userAgentSuffix != "" ? " " + dataTransferSettings.userAgentSuffix.get() : "")).c_str()); #if LIBCURL_VERSION_NUM >= 0x072b00 curl_easy_setopt(req, CURLOPT_PIPEWAIT, 1); #endif #if LIBCURL_VERSION_NUM >= 0x072f00 - if (downloadSettings.enableHttp2) + if (dataTransferSettings.enableHttp2) curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS); else curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1); @@ -297,10 +297,10 @@ struct CurlDownloader : public Downloader curl_easy_setopt(req, CURLOPT_SSL_VERIFYHOST, 0); } - curl_easy_setopt(req, CURLOPT_CONNECTTIMEOUT, downloadSettings.connectTimeout.get()); + curl_easy_setopt(req, CURLOPT_CONNECTTIMEOUT, dataTransferSettings.connectTimeout.get()); curl_easy_setopt(req, CURLOPT_LOW_SPEED_LIMIT, 1L); - curl_easy_setopt(req, CURLOPT_LOW_SPEED_TIME, downloadSettings.stalledDownloadTimeout.get()); + curl_easy_setopt(req, CURLOPT_LOW_SPEED_TIME, dataTransferSettings.stalledDownloadTimeout.get()); /* If no file exist in the specified path, curl continues to work anyway as if netrc support was disabled. */ @@ -469,7 +469,7 @@ struct CurlDownloader : public Downloader #endif #if LIBCURL_VERSION_NUM >= 0x071e00 // Max connections requires >= 7.30.0 curl_multi_setopt(curlm, CURLMOPT_MAX_TOTAL_CONNECTIONS, - downloadSettings.httpConnections.get()); + dataTransferSettings.httpConnections.get()); #endif wakeupPipe.create(); diff --git a/src/libstore/datatransfer.hh b/src/libstore/datatransfer.hh index 89d3fce7f..16955ff06 100644 --- a/src/libstore/datatransfer.hh +++ b/src/libstore/datatransfer.hh @@ -9,7 +9,7 @@ namespace nix { -struct DownloadSettings : Config +struct DataTransferSettings : Config { Setting enableHttp2{this, true, "http2", "Whether to enable HTTP/2 support."}; @@ -31,7 +31,7 @@ struct DownloadSettings : Config "How often Nix will attempt to download a file before giving up."}; }; -extern DownloadSettings downloadSettings; +extern DataTransferSettings dataTransferSettings; struct DataTransferRequest { @@ -39,7 +39,7 @@ struct DataTransferRequest std::string expectedETag; bool verifyTLS = true; bool head = false; - size_t tries = downloadSettings.tries; + size_t tries = dataTransferSettings.tries; unsigned int baseRetryTimeMs = 250; ActivityId parentAct; bool decompress = true; diff --git a/src/nix/main.cc b/src/nix/main.cc index ad4102201..6bb04d8b6 100644 --- a/src/nix/main.cc +++ b/src/nix/main.cc @@ -176,10 +176,10 @@ void mainWrapped(int argc, char * * argv) settings.useSubstitutes = false; if (!settings.tarballTtl.overriden) settings.tarballTtl = std::numeric_limits::max(); - if (!downloadSettings.tries.overriden) - downloadSettings.tries = 0; - if (!downloadSettings.connectTimeout.overriden) - downloadSettings.connectTimeout = 1; + if (!dataTransferSettings.tries.overriden) + dataTransferSettings.tries = 0; + if (!dataTransferSettings.connectTimeout.overriden) + dataTransferSettings.connectTimeout = 1; } if (args.refresh) -- cgit v1.2.3 From 142ed7fe45a2b1d566e4401b3974cb29b8944fc6 Mon Sep 17 00:00:00 2001 From: Nikola Knezevic Date: Mon, 6 Apr 2020 15:18:51 +0200 Subject: DownloadResult -> DataTransferResult --- src/libstore/datatransfer.cc | 20 ++++++++++---------- src/libstore/datatransfer.hh | 8 ++++---- src/libstore/http-binary-cache-store.cc | 2 +- src/libstore/s3-binary-cache-store.cc | 4 ++-- src/libstore/s3.hh | 4 ++-- 5 files changed, 19 insertions(+), 19 deletions(-) (limited to 'src') diff --git a/src/libstore/datatransfer.cc b/src/libstore/datatransfer.cc index 62f546cfa..06aa86d97 100644 --- a/src/libstore/datatransfer.cc +++ b/src/libstore/datatransfer.cc @@ -50,10 +50,10 @@ struct CurlDownloader : public Downloader { CurlDownloader & downloader; DataTransferRequest request; - DownloadResult result; + DataTransferResult result; Activity act; bool done = false; // whether either the success or failure function has been called - Callback callback; + Callback callback; CURL * req = 0; bool active = false; // whether the handle has been added to the multi object std::string status; @@ -74,7 +74,7 @@ struct CurlDownloader : public Downloader DownloadItem(CurlDownloader & downloader, const DataTransferRequest & request, - Callback && callback) + Callback && callback) : downloader(downloader) , request(request) , act(*logger, lvlTalkative, actDownload, @@ -642,7 +642,7 @@ struct CurlDownloader : public Downloader #endif void enqueueDownload(const DataTransferRequest & request, - Callback callback) override + Callback callback) override { /* Ugly hack to support s3:// URIs. */ if (hasPrefix(request.uri, "s3://")) { @@ -660,7 +660,7 @@ struct CurlDownloader : public Downloader // FIXME: implement ETag auto s3Res = s3Helper.getObject(bucketName, key); - DownloadResult res; + DataTransferResult res; if (!s3Res.data) throw DownloadError(NotFound, fmt("S3 object '%s' does not exist", request.uri)); res.data = s3Res.data; @@ -687,11 +687,11 @@ ref makeDownloader() return make_ref(); } -std::future Downloader::enqueueDownload(const DataTransferRequest & request) +std::future Downloader::enqueueDownload(const DataTransferRequest & request) { - auto promise = std::make_shared>(); + auto promise = std::make_shared>(); enqueueDownload(request, - {[promise](std::future fut) { + {[promise](std::future fut) { try { promise->set_value(fut.get()); } catch (...) { @@ -701,7 +701,7 @@ std::future Downloader::enqueueDownload(const DataTransferReques return promise->get_future(); } -DownloadResult Downloader::download(const DataTransferRequest & request) +DataTransferResult Downloader::download(const DataTransferRequest & request) { return enqueueDownload(request).get(); } @@ -756,7 +756,7 @@ void Downloader::download(DataTransferRequest && request, Sink & sink) }; enqueueDownload(request, - {[_state](std::future fut) { + {[_state](std::future fut) { auto state(_state->lock()); state->quit = true; try { diff --git a/src/libstore/datatransfer.hh b/src/libstore/datatransfer.hh index 16955ff06..b9e55655f 100644 --- a/src/libstore/datatransfer.hh +++ b/src/libstore/datatransfer.hh @@ -56,7 +56,7 @@ struct DataTransferRequest } }; -struct DownloadResult +struct DataTransferResult { bool cached = false; std::string etag; @@ -75,12 +75,12 @@ struct Downloader the download. The future may throw a DownloadError exception. */ virtual void enqueueDownload(const DataTransferRequest & request, - Callback callback) = 0; + Callback callback) = 0; - std::future enqueueDownload(const DataTransferRequest & request); + std::future enqueueDownload(const DataTransferRequest & request); /* Synchronously download a file. */ - DownloadResult download(const DataTransferRequest & request); + DataTransferResult download(const DataTransferRequest & request); /* Download a file, writing its data to a sink. The sink will be invoked on the thread of the caller. */ diff --git a/src/libstore/http-binary-cache-store.cc b/src/libstore/http-binary-cache-store.cc index 1fccee518..24a98d073 100644 --- a/src/libstore/http-binary-cache-store.cc +++ b/src/libstore/http-binary-cache-store.cc @@ -143,7 +143,7 @@ protected: auto callbackPtr = std::make_shared(std::move(callback)); getDownloader()->enqueueDownload(request, - {[callbackPtr, this](std::future result) { + {[callbackPtr, this](std::future result) { try { (*callbackPtr)(result.get().data); } catch (DownloadError & e) { diff --git a/src/libstore/s3-binary-cache-store.cc b/src/libstore/s3-binary-cache-store.cc index ed3c690b6..478cb9f84 100644 --- a/src/libstore/s3-binary-cache-store.cc +++ b/src/libstore/s3-binary-cache-store.cc @@ -132,7 +132,7 @@ ref S3Helper::makeConfig(const string & region return res; } -S3Helper::DownloadResult S3Helper::getObject( +S3Helper::DataTransferResult S3Helper::getObject( const std::string & bucketName, const std::string & key) { debug("fetching 's3://%s/%s'...", bucketName, key); @@ -146,7 +146,7 @@ S3Helper::DownloadResult S3Helper::getObject( return Aws::New("STRINGSTREAM"); }); - DownloadResult res; + DataTransferResult res; auto now1 = std::chrono::steady_clock::now(); diff --git a/src/libstore/s3.hh b/src/libstore/s3.hh index ef5f23d0f..d7d309243 100644 --- a/src/libstore/s3.hh +++ b/src/libstore/s3.hh @@ -18,13 +18,13 @@ struct S3Helper ref makeConfig(const std::string & region, const std::string & scheme, const std::string & endpoint); - struct DownloadResult + struct DataTransferResult { std::shared_ptr data; unsigned int durationMs; }; - DownloadResult getObject( + DataTransferResult getObject( const std::string & bucketName, const std::string & key); }; -- cgit v1.2.3 From 2df2741ec6e9411739b2049d7487b7b8a26d4547 Mon Sep 17 00:00:00 2001 From: Nikola Knezevic Date: Mon, 6 Apr 2020 15:37:33 +0200 Subject: enqueueDownload -> enqueueDataTransfer --- src/libstore/datatransfer.cc | 10 +++++----- src/libstore/datatransfer.hh | 4 ++-- src/libstore/http-binary-cache-store.cc | 2 +- 3 files changed, 8 insertions(+), 8 deletions(-) (limited to 'src') diff --git a/src/libstore/datatransfer.cc b/src/libstore/datatransfer.cc index 06aa86d97..9475c24ec 100644 --- a/src/libstore/datatransfer.cc +++ b/src/libstore/datatransfer.cc @@ -641,7 +641,7 @@ struct CurlDownloader : public Downloader } #endif - void enqueueDownload(const DataTransferRequest & request, + void enqueueDataTransfer(const DataTransferRequest & request, Callback callback) override { /* Ugly hack to support s3:// URIs. */ @@ -687,10 +687,10 @@ ref makeDownloader() return make_ref(); } -std::future Downloader::enqueueDownload(const DataTransferRequest & request) +std::future Downloader::enqueueDataTransfer(const DataTransferRequest & request) { auto promise = std::make_shared>(); - enqueueDownload(request, + enqueueDataTransfer(request, {[promise](std::future fut) { try { promise->set_value(fut.get()); @@ -703,7 +703,7 @@ std::future Downloader::enqueueDownload(const DataTransferRe DataTransferResult Downloader::download(const DataTransferRequest & request) { - return enqueueDownload(request).get(); + return enqueueDataTransfer(request).get(); } void Downloader::download(DataTransferRequest && request, Sink & sink) @@ -755,7 +755,7 @@ void Downloader::download(DataTransferRequest && request, Sink & sink) state->avail.notify_one(); }; - enqueueDownload(request, + enqueueDataTransfer(request, {[_state](std::future fut) { auto state(_state->lock()); state->quit = true; diff --git a/src/libstore/datatransfer.hh b/src/libstore/datatransfer.hh index b9e55655f..53159bdb4 100644 --- a/src/libstore/datatransfer.hh +++ b/src/libstore/datatransfer.hh @@ -74,10 +74,10 @@ struct Downloader /* Enqueue a download request, returning a future to the result of the download. The future may throw a DownloadError exception. */ - virtual void enqueueDownload(const DataTransferRequest & request, + virtual void enqueueDataTransfer(const DataTransferRequest & request, Callback callback) = 0; - std::future enqueueDownload(const DataTransferRequest & request); + std::future enqueueDataTransfer(const DataTransferRequest & request); /* Synchronously download a file. */ DataTransferResult download(const DataTransferRequest & request); diff --git a/src/libstore/http-binary-cache-store.cc b/src/libstore/http-binary-cache-store.cc index 24a98d073..861287a33 100644 --- a/src/libstore/http-binary-cache-store.cc +++ b/src/libstore/http-binary-cache-store.cc @@ -142,7 +142,7 @@ protected: auto callbackPtr = std::make_shared(std::move(callback)); - getDownloader()->enqueueDownload(request, + getDownloader()->enqueueDataTransfer(request, {[callbackPtr, this](std::future result) { try { (*callbackPtr)(result.get().data); -- cgit v1.2.3 From cd391206e6d29608bc3e25301b9921f1ac189086 Mon Sep 17 00:00:00 2001 From: Nikola Knezevic Date: Mon, 6 Apr 2020 22:57:09 +0200 Subject: {get,make,new}Downloader -> DataTransfer --- src/libstore/builtins/fetchurl.cc | 6 +++--- src/libstore/datatransfer.cc | 36 ++++++++++++++++---------------- src/libstore/datatransfer.hh | 18 ++++++++-------- src/libstore/http-binary-cache-store.cc | 14 ++++++------- src/nix-prefetch-url/nix-prefetch-url.cc | 2 +- src/nix/upgrade-nix.cc | 2 +- 6 files changed, 39 insertions(+), 39 deletions(-) (limited to 'src') diff --git a/src/libstore/builtins/fetchurl.cc b/src/libstore/builtins/fetchurl.cc index 01d21fa98..0b6db8f0c 100644 --- a/src/libstore/builtins/fetchurl.cc +++ b/src/libstore/builtins/fetchurl.cc @@ -26,9 +26,9 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData) auto mainUrl = getAttr("url"); bool unpack = get(drv.env, "unpack").value_or("") == "1"; - /* Note: have to use a fresh downloader here because we're in + /* Note: have to use a fresh dataTransfer here because we're in a forked process. */ - auto downloader = makeDownloader(); + auto dataTransfer = makeDataTransfer(); auto fetch = [&](const std::string & url) { @@ -42,7 +42,7 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData) auto decompressor = makeDecompressionSink( unpack && hasSuffix(mainUrl, ".xz") ? "xz" : "none", sink); - downloader->download(std::move(request), *decompressor); + dataTransfer->download(std::move(request), *decompressor); decompressor->finish(); }); diff --git a/src/libstore/datatransfer.cc b/src/libstore/datatransfer.cc index 9475c24ec..6ea9f9324 100644 --- a/src/libstore/datatransfer.cc +++ b/src/libstore/datatransfer.cc @@ -39,7 +39,7 @@ std::string resolveUri(const std::string & uri) return uri; } -struct CurlDownloader : public Downloader +struct curlDataTransfer : public DataTransfer { CURLM * curlm = 0; @@ -48,7 +48,7 @@ struct CurlDownloader : public Downloader struct DownloadItem : public std::enable_shared_from_this { - CurlDownloader & downloader; + curlDataTransfer & dataTransfer; DataTransferRequest request; DataTransferResult result; Activity act; @@ -72,10 +72,10 @@ struct CurlDownloader : public Downloader curl_off_t writtenToSink = 0; - DownloadItem(CurlDownloader & downloader, + DownloadItem(curlDataTransfer & dataTransfer, const DataTransferRequest & request, Callback && callback) - : downloader(downloader) + : dataTransfer(dataTransfer) , request(request) , act(*logger, lvlTalkative, actDownload, fmt(request.data ? "uploading '%s'" : "downloading '%s'", request.uri), @@ -106,7 +106,7 @@ struct CurlDownloader : public Downloader { if (req) { if (active) - curl_multi_remove_handle(downloader.curlm, req); + curl_multi_remove_handle(dataTransfer.curlm, req); curl_easy_cleanup(req); } if (requestHeaders) curl_slist_free_all(requestHeaders); @@ -422,13 +422,13 @@ struct CurlDownloader : public Downloader || writtenToSink == 0 || (acceptRanges && encoding.empty()))) { - int ms = request.baseRetryTimeMs * std::pow(2.0f, attempt - 1 + std::uniform_real_distribution<>(0.0, 0.5)(downloader.mt19937)); + int ms = request.baseRetryTimeMs * std::pow(2.0f, attempt - 1 + std::uniform_real_distribution<>(0.0, 0.5)(dataTransfer.mt19937)); if (writtenToSink) warn("%s; retrying from offset %d in %d ms", exc.what(), writtenToSink, ms); else warn("%s; retrying in %d ms", exc.what(), ms); embargo = std::chrono::steady_clock::now() + std::chrono::milliseconds(ms); - downloader.enqueueItem(shared_from_this()); + dataTransfer.enqueueItem(shared_from_this()); } else fail(exc); @@ -456,7 +456,7 @@ struct CurlDownloader : public Downloader std::thread workerThread; - CurlDownloader() + curlDataTransfer() : mt19937(rd()) { static std::once_flag globalInit; @@ -478,7 +478,7 @@ struct CurlDownloader : public Downloader workerThread = std::thread([&]() { workerThreadEntry(); }); } - ~CurlDownloader() + ~curlDataTransfer() { stopWorkerThread(); @@ -676,18 +676,18 @@ struct CurlDownloader : public Downloader } }; -ref getDownloader() +ref getDataTransfer() { - static ref downloader = makeDownloader(); - return downloader; + static ref dataTransfer = makeDataTransfer(); + return dataTransfer; } -ref makeDownloader() +ref makeDataTransfer() { - return make_ref(); + return make_ref(); } -std::future Downloader::enqueueDataTransfer(const DataTransferRequest & request) +std::future DataTransfer::enqueueDataTransfer(const DataTransferRequest & request) { auto promise = std::make_shared>(); enqueueDataTransfer(request, @@ -701,15 +701,15 @@ std::future Downloader::enqueueDataTransfer(const DataTransf return promise->get_future(); } -DataTransferResult Downloader::download(const DataTransferRequest & request) +DataTransferResult DataTransfer::download(const DataTransferRequest & request) { return enqueueDataTransfer(request).get(); } -void Downloader::download(DataTransferRequest && request, Sink & sink) +void DataTransfer::download(DataTransferRequest && request, Sink & sink) { /* Note: we can't call 'sink' via request.dataCallback, because - that would cause the sink to execute on the downloader + that would cause the sink to execute on the dataTransfer thread. If 'sink' is a coroutine, this will fail. Also, if the sink is expensive (e.g. one that does decompression and writing to the Nix store), it would stall the download thread too much. diff --git a/src/libstore/datatransfer.hh b/src/libstore/datatransfer.hh index 53159bdb4..e1c5196e0 100644 --- a/src/libstore/datatransfer.hh +++ b/src/libstore/datatransfer.hh @@ -67,11 +67,11 @@ struct DataTransferResult class Store; -struct Downloader +struct DataTransfer { - virtual ~Downloader() { } + virtual ~DataTransfer() { } - /* Enqueue a download request, returning a future to the result of + /* Enqueue a data transfer request, returning a future to the result of the download. The future may throw a DownloadError exception. */ virtual void enqueueDataTransfer(const DataTransferRequest & request, @@ -89,18 +89,18 @@ struct Downloader enum Error { NotFound, Forbidden, Misc, Transient, Interrupted }; }; -/* Return a shared Downloader object. Using this object is preferred +/* Return a shared DataTransfer object. Using this object is preferred because it enables connection reuse and HTTP/2 multiplexing. */ -ref getDownloader(); +ref getDataTransfer(); -/* Return a new Downloader object. */ -ref makeDownloader(); +/* Return a new DataTransfer object. */ +ref makeDataTransfer(); class DownloadError : public Error { public: - Downloader::Error error; - DownloadError(Downloader::Error error, const FormatOrString & fs) + DataTransfer::Error error; + DownloadError(DataTransfer::Error error, const FormatOrString & fs) : Error(fs), error(error) { } }; diff --git a/src/libstore/http-binary-cache-store.cc b/src/libstore/http-binary-cache-store.cc index 861287a33..9a19af5e0 100644 --- a/src/libstore/http-binary-cache-store.cc +++ b/src/libstore/http-binary-cache-store.cc @@ -87,12 +87,12 @@ protected: try { DataTransferRequest request(cacheUri + "/" + path); request.head = true; - getDownloader()->download(request); + getDataTransfer()->download(request); return true; } catch (DownloadError & e) { /* S3 buckets return 403 if a file doesn't exist and the bucket is unlistable, so treat 403 as 404. */ - if (e.error == Downloader::NotFound || e.error == Downloader::Forbidden) + if (e.error == DataTransfer::NotFound || e.error == DataTransfer::Forbidden) return false; maybeDisable(); throw; @@ -107,7 +107,7 @@ protected: req.data = std::make_shared(data); // FIXME: inefficient req.mimeType = mimeType; try { - getDownloader()->download(req); + getDataTransfer()->download(req); } catch (DownloadError & e) { throw UploadToHTTP("while uploading to HTTP binary cache at '%s': %s", cacheUri, e.msg()); } @@ -124,9 +124,9 @@ protected: checkEnabled(); auto request(makeRequest(path)); try { - getDownloader()->download(std::move(request), sink); + getDataTransfer()->download(std::move(request), sink); } catch (DownloadError & e) { - if (e.error == Downloader::NotFound || e.error == Downloader::Forbidden) + if (e.error == DataTransfer::NotFound || e.error == DataTransfer::Forbidden) throw NoSuchBinaryCacheFile("file '%s' does not exist in binary cache '%s'", path, getUri()); maybeDisable(); throw; @@ -142,12 +142,12 @@ protected: auto callbackPtr = std::make_shared(std::move(callback)); - getDownloader()->enqueueDataTransfer(request, + getDataTransfer()->enqueueDataTransfer(request, {[callbackPtr, this](std::future result) { try { (*callbackPtr)(result.get().data); } catch (DownloadError & e) { - if (e.error == Downloader::NotFound || e.error == Downloader::Forbidden) + if (e.error == DataTransfer::NotFound || e.error == DataTransfer::Forbidden) return (*callbackPtr)(std::shared_ptr()); maybeDisable(); callbackPtr->rethrow(); diff --git a/src/nix-prefetch-url/nix-prefetch-url.cc b/src/nix-prefetch-url/nix-prefetch-url.cc index 58a180e72..4c81e7d82 100644 --- a/src/nix-prefetch-url/nix-prefetch-url.cc +++ b/src/nix-prefetch-url/nix-prefetch-url.cc @@ -182,7 +182,7 @@ static int _main(int argc, char * * argv) DataTransferRequest req(actualUri); req.decompress = false; - getDownloader()->download(std::move(req), sink); + getDataTransfer()->download(std::move(req), sink); } /* Optionally unpack the file. */ diff --git a/src/nix/upgrade-nix.cc b/src/nix/upgrade-nix.cc index 414689ec5..a4eda90b4 100644 --- a/src/nix/upgrade-nix.cc +++ b/src/nix/upgrade-nix.cc @@ -139,7 +139,7 @@ struct CmdUpgradeNix : MixDryRun, StoreCommand // FIXME: use nixos.org? auto req = DataTransferRequest(storePathsUrl); - auto res = getDownloader()->download(req); + auto res = getDataTransfer()->download(req); auto state = std::make_unique(Strings(), store); auto v = state->allocValue(); -- cgit v1.2.3 From 213d124277460b316cfa03d43e4ec279bba86219 Mon Sep 17 00:00:00 2001 From: Nikola Knezevic Date: Mon, 6 Apr 2020 22:59:36 +0200 Subject: DownloadItem -> TransferItem --- src/libstore/datatransfer.cc | 32 ++++++++++++++++---------------- 1 file changed, 16 insertions(+), 16 deletions(-) (limited to 'src') diff --git a/src/libstore/datatransfer.cc b/src/libstore/datatransfer.cc index 6ea9f9324..452af107d 100644 --- a/src/libstore/datatransfer.cc +++ b/src/libstore/datatransfer.cc @@ -46,7 +46,7 @@ struct curlDataTransfer : public DataTransfer std::random_device rd; std::mt19937 mt19937; - struct DownloadItem : public std::enable_shared_from_this + struct TransferItem : public std::enable_shared_from_this { curlDataTransfer & dataTransfer; DataTransferRequest request; @@ -72,7 +72,7 @@ struct curlDataTransfer : public DataTransfer curl_off_t writtenToSink = 0; - DownloadItem(curlDataTransfer & dataTransfer, + TransferItem(curlDataTransfer & dataTransfer, const DataTransferRequest & request, Callback && callback) : dataTransfer(dataTransfer) @@ -102,7 +102,7 @@ struct curlDataTransfer : public DataTransfer requestHeaders = curl_slist_append(requestHeaders, ("Content-Type: " + request.mimeType).c_str()); } - ~DownloadItem() + ~TransferItem() { if (req) { if (active) @@ -156,7 +156,7 @@ struct curlDataTransfer : public DataTransfer static size_t writeCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) { - return ((DownloadItem *) userp)->writeCallback(contents, size, nmemb); + return ((TransferItem *) userp)->writeCallback(contents, size, nmemb); } size_t headerCallback(void * contents, size_t size, size_t nmemb) @@ -198,7 +198,7 @@ struct curlDataTransfer : public DataTransfer static size_t headerCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) { - return ((DownloadItem *) userp)->headerCallback(contents, size, nmemb); + return ((TransferItem *) userp)->headerCallback(contents, size, nmemb); } int progressCallback(double dltotal, double dlnow) @@ -213,7 +213,7 @@ struct curlDataTransfer : public DataTransfer static int progressCallbackWrapper(void * userp, double dltotal, double dlnow, double ultotal, double ulnow) { - return ((DownloadItem *) userp)->progressCallback(dltotal, dlnow); + return ((TransferItem *) userp)->progressCallback(dltotal, dlnow); } static int debugCallback(CURL * handle, curl_infotype type, char * data, size_t size, void * userptr) @@ -237,7 +237,7 @@ struct curlDataTransfer : public DataTransfer static size_t readCallbackWrapper(char *buffer, size_t size, size_t nitems, void * userp) { - return ((DownloadItem *) userp)->readCallback(buffer, size, nitems); + return ((TransferItem *) userp)->readCallback(buffer, size, nitems); } void init() @@ -248,7 +248,7 @@ struct curlDataTransfer : public DataTransfer if (verbosity >= lvlVomit) { curl_easy_setopt(req, CURLOPT_VERBOSE, 1); - curl_easy_setopt(req, CURLOPT_DEBUGFUNCTION, DownloadItem::debugCallback); + curl_easy_setopt(req, CURLOPT_DEBUGFUNCTION, TransferItem::debugCallback); } curl_easy_setopt(req, CURLOPT_URL, request.uri.c_str()); @@ -267,9 +267,9 @@ struct curlDataTransfer : public DataTransfer else curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1); #endif - curl_easy_setopt(req, CURLOPT_WRITEFUNCTION, DownloadItem::writeCallbackWrapper); + curl_easy_setopt(req, CURLOPT_WRITEFUNCTION, TransferItem::writeCallbackWrapper); curl_easy_setopt(req, CURLOPT_WRITEDATA, this); - curl_easy_setopt(req, CURLOPT_HEADERFUNCTION, DownloadItem::headerCallbackWrapper); + curl_easy_setopt(req, CURLOPT_HEADERFUNCTION, TransferItem::headerCallbackWrapper); curl_easy_setopt(req, CURLOPT_HEADERDATA, this); curl_easy_setopt(req, CURLOPT_PROGRESSFUNCTION, progressCallbackWrapper); @@ -439,12 +439,12 @@ struct curlDataTransfer : public DataTransfer struct State { struct EmbargoComparator { - bool operator() (const std::shared_ptr & i1, const std::shared_ptr & i2) { + bool operator() (const std::shared_ptr & i1, const std::shared_ptr & i2) { return i1->embargo > i2->embargo; } }; bool quit = false; - std::priority_queue, std::vector>, EmbargoComparator> incoming; + std::priority_queue, std::vector>, EmbargoComparator> incoming; }; Sync state_; @@ -504,7 +504,7 @@ struct curlDataTransfer : public DataTransfer stopWorkerThread(); }); - std::map> items; + std::map> items; bool quit = false; @@ -561,7 +561,7 @@ struct curlDataTransfer : public DataTransfer throw SysError("reading curl wakeup socket"); } - std::vector> incoming; + std::vector> incoming; auto now = std::chrono::steady_clock::now(); { @@ -609,7 +609,7 @@ struct curlDataTransfer : public DataTransfer } } - void enqueueItem(std::shared_ptr item) + void enqueueItem(std::shared_ptr item) { if (item->request.data && !hasPrefix(item->request.uri, "http://") @@ -672,7 +672,7 @@ struct curlDataTransfer : public DataTransfer return; } - enqueueItem(std::make_shared(*this, request, std::move(callback))); + enqueueItem(std::make_shared(*this, request, std::move(callback))); } }; -- cgit v1.2.3 From c4c1ae0a00e8dd3258af58c15b664828b592133e Mon Sep 17 00:00:00 2001 From: Nikola Knezevic Date: Mon, 6 Apr 2020 23:00:43 +0200 Subject: DownloadError -> DataTransferError --- src/libexpr/parser.y | 2 +- src/libstore/datatransfer.cc | 10 +++++----- src/libstore/datatransfer.hh | 6 +++--- src/libstore/http-binary-cache-store.cc | 8 ++++---- src/nix-channel/nix-channel.cc | 2 +- 5 files changed, 14 insertions(+), 14 deletions(-) (limited to 'src') diff --git a/src/libexpr/parser.y b/src/libexpr/parser.y index 96b6f0ecd..8fbe59301 100644 --- a/src/libexpr/parser.y +++ b/src/libexpr/parser.y @@ -690,7 +690,7 @@ std::pair EvalState::resolveSearchPathElem(const SearchPathEl try { res = { true, store->toRealPath(fetchers::downloadTarball( store, resolveUri(elem.second), "source", false).storePath) }; - } catch (DownloadError & e) { + } catch (DataTransferError & e) { printError(format("warning: Nix search path entry '%1%' cannot be downloaded, ignoring") % elem.second); res = { false, "" }; } diff --git a/src/libstore/datatransfer.cc b/src/libstore/datatransfer.cc index 452af107d..be5dd27cd 100644 --- a/src/libstore/datatransfer.cc +++ b/src/libstore/datatransfer.cc @@ -112,7 +112,7 @@ struct curlDataTransfer : public DataTransfer if (requestHeaders) curl_slist_free_all(requestHeaders); try { if (!done) - fail(DownloadError(Interrupted, format("download of '%s' was interrupted") % request.uri)); + fail(DataTransferError(Interrupted, format("download of '%s' was interrupted") % request.uri)); } catch (...) { ignoreException(); } @@ -401,14 +401,14 @@ struct curlDataTransfer : public DataTransfer auto exc = code == CURLE_ABORTED_BY_CALLBACK && _isInterrupted - ? DownloadError(Interrupted, fmt("%s of '%s' was interrupted", request.verb(), request.uri)) + ? DataTransferError(Interrupted, fmt("%s of '%s' was interrupted", request.verb(), request.uri)) : httpStatus != 0 - ? DownloadError(err, + ? DataTransferError(err, fmt("unable to %s '%s': HTTP error %d", request.verb(), request.uri, httpStatus) + (code == CURLE_OK ? "" : fmt(" (curl error: %s)", curl_easy_strerror(code))) ) - : DownloadError(err, + : DataTransferError(err, fmt("unable to %s '%s': %s (%d)", request.verb(), request.uri, curl_easy_strerror(code), code)); @@ -662,7 +662,7 @@ struct curlDataTransfer : public DataTransfer auto s3Res = s3Helper.getObject(bucketName, key); DataTransferResult res; if (!s3Res.data) - throw DownloadError(NotFound, fmt("S3 object '%s' does not exist", request.uri)); + throw DataTransferError(NotFound, fmt("S3 object '%s' does not exist", request.uri)); res.data = s3Res.data; callback(std::move(res)); #else diff --git a/src/libstore/datatransfer.hh b/src/libstore/datatransfer.hh index e1c5196e0..f2ea53ae3 100644 --- a/src/libstore/datatransfer.hh +++ b/src/libstore/datatransfer.hh @@ -72,7 +72,7 @@ struct DataTransfer virtual ~DataTransfer() { } /* Enqueue a data transfer request, returning a future to the result of - the download. The future may throw a DownloadError + the download. The future may throw a DataTransferError exception. */ virtual void enqueueDataTransfer(const DataTransferRequest & request, Callback callback) = 0; @@ -96,11 +96,11 @@ ref getDataTransfer(); /* Return a new DataTransfer object. */ ref makeDataTransfer(); -class DownloadError : public Error +class DataTransferError : public Error { public: DataTransfer::Error error; - DownloadError(DataTransfer::Error error, const FormatOrString & fs) + DataTransferError(DataTransfer::Error error, const FormatOrString & fs) : Error(fs), error(error) { } }; diff --git a/src/libstore/http-binary-cache-store.cc b/src/libstore/http-binary-cache-store.cc index 9a19af5e0..5e446ddf2 100644 --- a/src/libstore/http-binary-cache-store.cc +++ b/src/libstore/http-binary-cache-store.cc @@ -89,7 +89,7 @@ protected: request.head = true; getDataTransfer()->download(request); return true; - } catch (DownloadError & e) { + } catch (DataTransferError & e) { /* S3 buckets return 403 if a file doesn't exist and the bucket is unlistable, so treat 403 as 404. */ if (e.error == DataTransfer::NotFound || e.error == DataTransfer::Forbidden) @@ -108,7 +108,7 @@ protected: req.mimeType = mimeType; try { getDataTransfer()->download(req); - } catch (DownloadError & e) { + } catch (DataTransferError & e) { throw UploadToHTTP("while uploading to HTTP binary cache at '%s': %s", cacheUri, e.msg()); } } @@ -125,7 +125,7 @@ protected: auto request(makeRequest(path)); try { getDataTransfer()->download(std::move(request), sink); - } catch (DownloadError & e) { + } catch (DataTransferError & e) { if (e.error == DataTransfer::NotFound || e.error == DataTransfer::Forbidden) throw NoSuchBinaryCacheFile("file '%s' does not exist in binary cache '%s'", path, getUri()); maybeDisable(); @@ -146,7 +146,7 @@ protected: {[callbackPtr, this](std::future result) { try { (*callbackPtr)(result.get().data); - } catch (DownloadError & e) { + } catch (DataTransferError & e) { if (e.error == DataTransfer::NotFound || e.error == DataTransfer::Forbidden) return (*callbackPtr)(std::shared_ptr()); maybeDisable(); diff --git a/src/nix-channel/nix-channel.cc b/src/nix-channel/nix-channel.cc index d0719194d..0f8024d7b 100755 --- a/src/nix-channel/nix-channel.cc +++ b/src/nix-channel/nix-channel.cc @@ -113,7 +113,7 @@ static void update(const StringSet & channelNames) // Download the channel tarball. try { filename = store->toRealPath(fetchers::downloadFile(store, url + "/nixexprs.tar.xz", "nixexprs.tar.xz", false).storePath); - } catch (DownloadError & e) { + } catch (DataTransferError & e) { filename = store->toRealPath(fetchers::downloadFile(store, url + "/nixexprs.tar.bz2", "nixexprs.tar.bz2", false).storePath); } } -- cgit v1.2.3 From a0c5931208042da39bb6a5e80a4b27cf50f665d6 Mon Sep 17 00:00:00 2001 From: Nikola Knezevic Date: Mon, 6 Apr 2020 23:08:39 +0200 Subject: actDownload -> actDataTransfer --- src/libstore/build.cc | 2 +- src/libstore/datatransfer.cc | 2 +- src/libutil/logging.cc | 2 +- src/libutil/logging.hh | 2 +- src/nix/progress-bar.cc | 6 +++--- 5 files changed, 7 insertions(+), 7 deletions(-) (limited to 'src') diff --git a/src/libstore/build.cc b/src/libstore/build.cc index db01d9510..f091a000d 100644 --- a/src/libstore/build.cc +++ b/src/libstore/build.cc @@ -361,7 +361,7 @@ public: { actDerivations.progress(doneBuilds, expectedBuilds + doneBuilds, runningBuilds, failedBuilds); actSubstitutions.progress(doneSubstitutions, expectedSubstitutions + doneSubstitutions, runningSubstitutions, failedSubstitutions); - act.setExpected(actDownload, expectedDownloadSize + doneDownloadSize); + act.setExpected(actDataTransfer, expectedDownloadSize + doneDownloadSize); act.setExpected(actCopyPath, expectedNarSize + doneNarSize); } }; diff --git a/src/libstore/datatransfer.cc b/src/libstore/datatransfer.cc index be5dd27cd..9ed06b5c1 100644 --- a/src/libstore/datatransfer.cc +++ b/src/libstore/datatransfer.cc @@ -77,7 +77,7 @@ struct curlDataTransfer : public DataTransfer Callback && callback) : dataTransfer(dataTransfer) , request(request) - , act(*logger, lvlTalkative, actDownload, + , act(*logger, lvlTalkative, actDataTransfer, fmt(request.data ? "uploading '%s'" : "downloading '%s'", request.uri), {request.uri}, request.parentAct) , callback(std::move(callback)) diff --git a/src/libutil/logging.cc b/src/libutil/logging.cc index fa5c84a27..001a3fb28 100644 --- a/src/libutil/logging.cc +++ b/src/libutil/logging.cc @@ -198,7 +198,7 @@ bool handleJSONLogMessage(const std::string & msg, if (action == "start") { auto type = (ActivityType) json["type"]; - if (trusted || type == actDownload) + if (trusted || type == actDataTransfer) activities.emplace(std::piecewise_construct, std::forward_as_tuple(json["id"]), std::forward_as_tuple(*logger, (Verbosity) json["level"], type, diff --git a/src/libutil/logging.hh b/src/libutil/logging.hh index beb5e6b64..423ed95da 100644 --- a/src/libutil/logging.hh +++ b/src/libutil/logging.hh @@ -17,7 +17,7 @@ typedef enum { typedef enum { actUnknown = 0, actCopyPath = 100, - actDownload = 101, + actDataTransfer = 101, actRealise = 102, actCopyPaths = 103, actBuilds = 104, diff --git a/src/nix/progress-bar.cc b/src/nix/progress-bar.cc index 26631416c..d108896ad 100644 --- a/src/nix/progress-bar.cc +++ b/src/nix/progress-bar.cc @@ -190,8 +190,8 @@ public: i->s = fmt("querying " ANSI_BOLD "%s" ANSI_NORMAL " on %s", name, getS(fields, 1)); } - if ((type == actDownload && hasAncestor(*state, actCopyPath, parent)) - || (type == actDownload && hasAncestor(*state, actQueryPathInfo, parent)) + if ((type == actDataTransfer && hasAncestor(*state, actCopyPath, parent)) + || (type == actDataTransfer && hasAncestor(*state, actQueryPathInfo, parent)) || (type == actCopyPath && hasAncestor(*state, actSubstitute, parent))) i->visible = false; @@ -416,7 +416,7 @@ public: if (!s2.empty()) { res += " ("; res += s2; res += ')'; } } - showActivity(actDownload, "%s MiB DL", "%.1f", MiB); + showActivity(actDataTransfer, "%s MiB DL", "%.1f", MiB); { auto s = renderActivity(actOptimiseStore, "%s paths optimised"); -- cgit v1.2.3 From 7848372b0f683a5a4db5f86fd998e8df5fa22715 Mon Sep 17 00:00:00 2001 From: Nikola Knezevic Date: Mon, 6 Apr 2020 23:34:31 +0200 Subject: Add upload method --- src/libstore/datatransfer.cc | 6 ++++++ src/libstore/datatransfer.hh | 3 +++ src/libstore/http-binary-cache-store.cc | 3 +-- 3 files changed, 10 insertions(+), 2 deletions(-) (limited to 'src') diff --git a/src/libstore/datatransfer.cc b/src/libstore/datatransfer.cc index 9ed06b5c1..dbdf07ad4 100644 --- a/src/libstore/datatransfer.cc +++ b/src/libstore/datatransfer.cc @@ -706,6 +706,12 @@ DataTransferResult DataTransfer::download(const DataTransferRequest & request) return enqueueDataTransfer(request).get(); } +DataTransferResult DataTransfer::upload(const DataTransferRequest & request) +{ + /* Note: this method is the same as download, but helps in readability */ + return enqueueDataTransfer(request).get(); +} + void DataTransfer::download(DataTransferRequest && request, Sink & sink) { /* Note: we can't call 'sink' via request.dataCallback, because diff --git a/src/libstore/datatransfer.hh b/src/libstore/datatransfer.hh index f2ea53ae3..68d97ceb1 100644 --- a/src/libstore/datatransfer.hh +++ b/src/libstore/datatransfer.hh @@ -82,6 +82,9 @@ struct DataTransfer /* Synchronously download a file. */ DataTransferResult download(const DataTransferRequest & request); + /* Synchronously upload a file. */ + DataTransferResult upload(const DataTransferRequest & request); + /* Download a file, writing its data to a sink. The sink will be invoked on the thread of the caller. */ void download(DataTransferRequest && request, Sink & sink); diff --git a/src/libstore/http-binary-cache-store.cc b/src/libstore/http-binary-cache-store.cc index 5e446ddf2..c75846bc3 100644 --- a/src/libstore/http-binary-cache-store.cc +++ b/src/libstore/http-binary-cache-store.cc @@ -107,7 +107,7 @@ protected: req.data = std::make_shared(data); // FIXME: inefficient req.mimeType = mimeType; try { - getDataTransfer()->download(req); + getDataTransfer()->upload(req); } catch (DataTransferError & e) { throw UploadToHTTP("while uploading to HTTP binary cache at '%s': %s", cacheUri, e.msg()); } @@ -174,4 +174,3 @@ static RegisterStoreImplementation regStore([]( }); } - -- cgit v1.2.3 From c330109bfa38370f7cef6449efb88309a78e1684 Mon Sep 17 00:00:00 2001 From: Nikola Knezevic Date: Mon, 6 Apr 2020 23:43:43 +0200 Subject: DataTransfer -> FileTransfer --- src/libstore/build.cc | 2 +- src/libstore/builtins/fetchurl.cc | 8 +-- src/libstore/datatransfer.cc | 94 ++++++++++++++++---------------- src/libstore/datatransfer.hh | 44 +++++++-------- src/libstore/http-binary-cache-store.cc | 32 +++++------ src/libstore/s3-binary-cache-store.cc | 4 +- src/libstore/s3.hh | 4 +- src/libutil/logging.cc | 2 +- src/libutil/logging.hh | 2 +- src/nix-prefetch-url/nix-prefetch-url.cc | 4 +- src/nix/main.cc | 8 +-- src/nix/progress-bar.cc | 6 +- src/nix/upgrade-nix.cc | 4 +- 13 files changed, 107 insertions(+), 107 deletions(-) (limited to 'src') diff --git a/src/libstore/build.cc b/src/libstore/build.cc index f091a000d..9baa2cf1a 100644 --- a/src/libstore/build.cc +++ b/src/libstore/build.cc @@ -361,7 +361,7 @@ public: { actDerivations.progress(doneBuilds, expectedBuilds + doneBuilds, runningBuilds, failedBuilds); actSubstitutions.progress(doneSubstitutions, expectedSubstitutions + doneSubstitutions, runningSubstitutions, failedSubstitutions); - act.setExpected(actDataTransfer, expectedDownloadSize + doneDownloadSize); + act.setExpected(actFileTransfer, expectedDownloadSize + doneDownloadSize); act.setExpected(actCopyPath, expectedNarSize + doneNarSize); } }; diff --git a/src/libstore/builtins/fetchurl.cc b/src/libstore/builtins/fetchurl.cc index 0b6db8f0c..1432de9d7 100644 --- a/src/libstore/builtins/fetchurl.cc +++ b/src/libstore/builtins/fetchurl.cc @@ -26,9 +26,9 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData) auto mainUrl = getAttr("url"); bool unpack = get(drv.env, "unpack").value_or("") == "1"; - /* Note: have to use a fresh dataTransfer here because we're in + /* Note: have to use a fresh fileTransfer here because we're in a forked process. */ - auto dataTransfer = makeDataTransfer(); + auto fileTransfer = makeFileTransfer(); auto fetch = [&](const std::string & url) { @@ -36,13 +36,13 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData) /* No need to do TLS verification, because we check the hash of the result anyway. */ - DataTransferRequest request(url); + FileTransferRequest request(url); request.verifyTLS = false; request.decompress = false; auto decompressor = makeDecompressionSink( unpack && hasSuffix(mainUrl, ".xz") ? "xz" : "none", sink); - dataTransfer->download(std::move(request), *decompressor); + fileTransfer->download(std::move(request), *decompressor); decompressor->finish(); }); diff --git a/src/libstore/datatransfer.cc b/src/libstore/datatransfer.cc index dbdf07ad4..5cc63a1d8 100644 --- a/src/libstore/datatransfer.cc +++ b/src/libstore/datatransfer.cc @@ -27,9 +27,9 @@ using namespace std::string_literals; namespace nix { -DataTransferSettings dataTransferSettings; +FileTransferSettings fileTransferSettings; -static GlobalConfig::Register r1(&dataTransferSettings); +static GlobalConfig::Register r1(&fileTransferSettings); std::string resolveUri(const std::string & uri) { @@ -39,7 +39,7 @@ std::string resolveUri(const std::string & uri) return uri; } -struct curlDataTransfer : public DataTransfer +struct curlFileTransfer : public FileTransfer { CURLM * curlm = 0; @@ -48,12 +48,12 @@ struct curlDataTransfer : public DataTransfer struct TransferItem : public std::enable_shared_from_this { - curlDataTransfer & dataTransfer; - DataTransferRequest request; - DataTransferResult result; + curlFileTransfer & fileTransfer; + FileTransferRequest request; + FileTransferResult result; Activity act; bool done = false; // whether either the success or failure function has been called - Callback callback; + Callback callback; CURL * req = 0; bool active = false; // whether the handle has been added to the multi object std::string status; @@ -72,12 +72,12 @@ struct curlDataTransfer : public DataTransfer curl_off_t writtenToSink = 0; - TransferItem(curlDataTransfer & dataTransfer, - const DataTransferRequest & request, - Callback && callback) - : dataTransfer(dataTransfer) + TransferItem(curlFileTransfer & fileTransfer, + const FileTransferRequest & request, + Callback && callback) + : fileTransfer(fileTransfer) , request(request) - , act(*logger, lvlTalkative, actDataTransfer, + , act(*logger, lvlTalkative, actFileTransfer, fmt(request.data ? "uploading '%s'" : "downloading '%s'", request.uri), {request.uri}, request.parentAct) , callback(std::move(callback)) @@ -106,13 +106,13 @@ struct curlDataTransfer : public DataTransfer { if (req) { if (active) - curl_multi_remove_handle(dataTransfer.curlm, req); + curl_multi_remove_handle(fileTransfer.curlm, req); curl_easy_cleanup(req); } if (requestHeaders) curl_slist_free_all(requestHeaders); try { if (!done) - fail(DataTransferError(Interrupted, format("download of '%s' was interrupted") % request.uri)); + fail(FileTransferError(Interrupted, format("download of '%s' was interrupted") % request.uri)); } catch (...) { ignoreException(); } @@ -257,12 +257,12 @@ struct curlDataTransfer : public DataTransfer curl_easy_setopt(req, CURLOPT_NOSIGNAL, 1); curl_easy_setopt(req, CURLOPT_USERAGENT, ("curl/" LIBCURL_VERSION " Nix/" + nixVersion + - (dataTransferSettings.userAgentSuffix != "" ? " " + dataTransferSettings.userAgentSuffix.get() : "")).c_str()); + (fileTransferSettings.userAgentSuffix != "" ? " " + fileTransferSettings.userAgentSuffix.get() : "")).c_str()); #if LIBCURL_VERSION_NUM >= 0x072b00 curl_easy_setopt(req, CURLOPT_PIPEWAIT, 1); #endif #if LIBCURL_VERSION_NUM >= 0x072f00 - if (dataTransferSettings.enableHttp2) + if (fileTransferSettings.enableHttp2) curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS); else curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1); @@ -297,10 +297,10 @@ struct curlDataTransfer : public DataTransfer curl_easy_setopt(req, CURLOPT_SSL_VERIFYHOST, 0); } - curl_easy_setopt(req, CURLOPT_CONNECTTIMEOUT, dataTransferSettings.connectTimeout.get()); + curl_easy_setopt(req, CURLOPT_CONNECTTIMEOUT, fileTransferSettings.connectTimeout.get()); curl_easy_setopt(req, CURLOPT_LOW_SPEED_LIMIT, 1L); - curl_easy_setopt(req, CURLOPT_LOW_SPEED_TIME, dataTransferSettings.stalledDownloadTimeout.get()); + curl_easy_setopt(req, CURLOPT_LOW_SPEED_TIME, fileTransferSettings.stalledDownloadTimeout.get()); /* If no file exist in the specified path, curl continues to work anyway as if netrc support was disabled. */ @@ -401,14 +401,14 @@ struct curlDataTransfer : public DataTransfer auto exc = code == CURLE_ABORTED_BY_CALLBACK && _isInterrupted - ? DataTransferError(Interrupted, fmt("%s of '%s' was interrupted", request.verb(), request.uri)) + ? FileTransferError(Interrupted, fmt("%s of '%s' was interrupted", request.verb(), request.uri)) : httpStatus != 0 - ? DataTransferError(err, + ? FileTransferError(err, fmt("unable to %s '%s': HTTP error %d", request.verb(), request.uri, httpStatus) + (code == CURLE_OK ? "" : fmt(" (curl error: %s)", curl_easy_strerror(code))) ) - : DataTransferError(err, + : FileTransferError(err, fmt("unable to %s '%s': %s (%d)", request.verb(), request.uri, curl_easy_strerror(code), code)); @@ -422,13 +422,13 @@ struct curlDataTransfer : public DataTransfer || writtenToSink == 0 || (acceptRanges && encoding.empty()))) { - int ms = request.baseRetryTimeMs * std::pow(2.0f, attempt - 1 + std::uniform_real_distribution<>(0.0, 0.5)(dataTransfer.mt19937)); + int ms = request.baseRetryTimeMs * std::pow(2.0f, attempt - 1 + std::uniform_real_distribution<>(0.0, 0.5)(fileTransfer.mt19937)); if (writtenToSink) warn("%s; retrying from offset %d in %d ms", exc.what(), writtenToSink, ms); else warn("%s; retrying in %d ms", exc.what(), ms); embargo = std::chrono::steady_clock::now() + std::chrono::milliseconds(ms); - dataTransfer.enqueueItem(shared_from_this()); + fileTransfer.enqueueItem(shared_from_this()); } else fail(exc); @@ -456,7 +456,7 @@ struct curlDataTransfer : public DataTransfer std::thread workerThread; - curlDataTransfer() + curlFileTransfer() : mt19937(rd()) { static std::once_flag globalInit; @@ -469,7 +469,7 @@ struct curlDataTransfer : public DataTransfer #endif #if LIBCURL_VERSION_NUM >= 0x071e00 // Max connections requires >= 7.30.0 curl_multi_setopt(curlm, CURLMOPT_MAX_TOTAL_CONNECTIONS, - dataTransferSettings.httpConnections.get()); + fileTransferSettings.httpConnections.get()); #endif wakeupPipe.create(); @@ -478,7 +478,7 @@ struct curlDataTransfer : public DataTransfer workerThread = std::thread([&]() { workerThreadEntry(); }); } - ~curlDataTransfer() + ~curlFileTransfer() { stopWorkerThread(); @@ -641,8 +641,8 @@ struct curlDataTransfer : public DataTransfer } #endif - void enqueueDataTransfer(const DataTransferRequest & request, - Callback callback) override + void enqueueFileTransfer(const FileTransferRequest & request, + Callback callback) override { /* Ugly hack to support s3:// URIs. */ if (hasPrefix(request.uri, "s3://")) { @@ -660,9 +660,9 @@ struct curlDataTransfer : public DataTransfer // FIXME: implement ETag auto s3Res = s3Helper.getObject(bucketName, key); - DataTransferResult res; + FileTransferResult res; if (!s3Res.data) - throw DataTransferError(NotFound, fmt("S3 object '%s' does not exist", request.uri)); + throw FileTransferError(NotFound, fmt("S3 object '%s' does not exist", request.uri)); res.data = s3Res.data; callback(std::move(res)); #else @@ -676,22 +676,22 @@ struct curlDataTransfer : public DataTransfer } }; -ref getDataTransfer() +ref getFileTransfer() { - static ref dataTransfer = makeDataTransfer(); - return dataTransfer; + static ref fileTransfer = makeFileTransfer(); + return fileTransfer; } -ref makeDataTransfer() +ref makeFileTransfer() { - return make_ref(); + return make_ref(); } -std::future DataTransfer::enqueueDataTransfer(const DataTransferRequest & request) +std::future FileTransfer::enqueueFileTransfer(const FileTransferRequest & request) { - auto promise = std::make_shared>(); - enqueueDataTransfer(request, - {[promise](std::future fut) { + auto promise = std::make_shared>(); + enqueueFileTransfer(request, + {[promise](std::future fut) { try { promise->set_value(fut.get()); } catch (...) { @@ -701,21 +701,21 @@ std::future DataTransfer::enqueueDataTransfer(const DataTran return promise->get_future(); } -DataTransferResult DataTransfer::download(const DataTransferRequest & request) +FileTransferResult FileTransfer::download(const FileTransferRequest & request) { - return enqueueDataTransfer(request).get(); + return enqueueFileTransfer(request).get(); } -DataTransferResult DataTransfer::upload(const DataTransferRequest & request) +FileTransferResult FileTransfer::upload(const FileTransferRequest & request) { /* Note: this method is the same as download, but helps in readability */ - return enqueueDataTransfer(request).get(); + return enqueueFileTransfer(request).get(); } -void DataTransfer::download(DataTransferRequest && request, Sink & sink) +void FileTransfer::download(FileTransferRequest && request, Sink & sink) { /* Note: we can't call 'sink' via request.dataCallback, because - that would cause the sink to execute on the dataTransfer + that would cause the sink to execute on the fileTransfer thread. If 'sink' is a coroutine, this will fail. Also, if the sink is expensive (e.g. one that does decompression and writing to the Nix store), it would stall the download thread too much. @@ -761,8 +761,8 @@ void DataTransfer::download(DataTransferRequest && request, Sink & sink) state->avail.notify_one(); }; - enqueueDataTransfer(request, - {[_state](std::future fut) { + enqueueFileTransfer(request, + {[_state](std::future fut) { auto state(_state->lock()); state->quit = true; try { diff --git a/src/libstore/datatransfer.hh b/src/libstore/datatransfer.hh index 68d97ceb1..2347f363d 100644 --- a/src/libstore/datatransfer.hh +++ b/src/libstore/datatransfer.hh @@ -9,7 +9,7 @@ namespace nix { -struct DataTransferSettings : Config +struct FileTransferSettings : Config { Setting enableHttp2{this, true, "http2", "Whether to enable HTTP/2 support."}; @@ -31,15 +31,15 @@ struct DataTransferSettings : Config "How often Nix will attempt to download a file before giving up."}; }; -extern DataTransferSettings dataTransferSettings; +extern FileTransferSettings fileTransferSettings; -struct DataTransferRequest +struct FileTransferRequest { std::string uri; std::string expectedETag; bool verifyTLS = true; bool head = false; - size_t tries = dataTransferSettings.tries; + size_t tries = fileTransferSettings.tries; unsigned int baseRetryTimeMs = 250; ActivityId parentAct; bool decompress = true; @@ -47,7 +47,7 @@ struct DataTransferRequest std::string mimeType; std::function dataCallback; - DataTransferRequest(const std::string & uri) + FileTransferRequest(const std::string & uri) : uri(uri), parentAct(getCurActivity()) { } std::string verb() @@ -56,7 +56,7 @@ struct DataTransferRequest } }; -struct DataTransferResult +struct FileTransferResult { bool cached = false; std::string etag; @@ -67,43 +67,43 @@ struct DataTransferResult class Store; -struct DataTransfer +struct FileTransfer { - virtual ~DataTransfer() { } + virtual ~FileTransfer() { } /* Enqueue a data transfer request, returning a future to the result of - the download. The future may throw a DataTransferError + the download. The future may throw a FileTransferError exception. */ - virtual void enqueueDataTransfer(const DataTransferRequest & request, - Callback callback) = 0; + virtual void enqueueFileTransfer(const FileTransferRequest & request, + Callback callback) = 0; - std::future enqueueDataTransfer(const DataTransferRequest & request); + std::future enqueueFileTransfer(const FileTransferRequest & request); /* Synchronously download a file. */ - DataTransferResult download(const DataTransferRequest & request); + FileTransferResult download(const FileTransferRequest & request); /* Synchronously upload a file. */ - DataTransferResult upload(const DataTransferRequest & request); + FileTransferResult upload(const FileTransferRequest & request); /* Download a file, writing its data to a sink. The sink will be invoked on the thread of the caller. */ - void download(DataTransferRequest && request, Sink & sink); + void download(FileTransferRequest && request, Sink & sink); enum Error { NotFound, Forbidden, Misc, Transient, Interrupted }; }; -/* Return a shared DataTransfer object. Using this object is preferred +/* Return a shared FileTransfer object. Using this object is preferred because it enables connection reuse and HTTP/2 multiplexing. */ -ref getDataTransfer(); +ref getFileTransfer(); -/* Return a new DataTransfer object. */ -ref makeDataTransfer(); +/* Return a new FileTransfer object. */ +ref makeFileTransfer(); -class DataTransferError : public Error +class FileTransferError : public Error { public: - DataTransfer::Error error; - DataTransferError(DataTransfer::Error error, const FormatOrString & fs) + FileTransfer::Error error; + FileTransferError(FileTransfer::Error error, const FormatOrString & fs) : Error(fs), error(error) { } }; diff --git a/src/libstore/http-binary-cache-store.cc b/src/libstore/http-binary-cache-store.cc index c75846bc3..ce0c5af3b 100644 --- a/src/libstore/http-binary-cache-store.cc +++ b/src/libstore/http-binary-cache-store.cc @@ -85,14 +85,14 @@ protected: checkEnabled(); try { - DataTransferRequest request(cacheUri + "/" + path); + FileTransferRequest request(cacheUri + "/" + path); request.head = true; - getDataTransfer()->download(request); + getFileTransfer()->download(request); return true; - } catch (DataTransferError & e) { + } 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 == DataTransfer::NotFound || e.error == DataTransfer::Forbidden) + if (e.error == FileTransfer::NotFound || e.error == FileTransfer::Forbidden) return false; maybeDisable(); throw; @@ -103,19 +103,19 @@ protected: const std::string & data, const std::string & mimeType) override { - auto req = DataTransferRequest(cacheUri + "/" + path); + auto req = FileTransferRequest(cacheUri + "/" + path); req.data = std::make_shared(data); // FIXME: inefficient req.mimeType = mimeType; try { - getDataTransfer()->upload(req); - } catch (DataTransferError & e) { + getFileTransfer()->upload(req); + } catch (FileTransferError & e) { throw UploadToHTTP("while uploading to HTTP binary cache at '%s': %s", cacheUri, e.msg()); } } - DataTransferRequest makeRequest(const std::string & path) + FileTransferRequest makeRequest(const std::string & path) { - DataTransferRequest request(cacheUri + "/" + path); + FileTransferRequest request(cacheUri + "/" + path); return request; } @@ -124,9 +124,9 @@ protected: checkEnabled(); auto request(makeRequest(path)); try { - getDataTransfer()->download(std::move(request), sink); - } catch (DataTransferError & e) { - if (e.error == DataTransfer::NotFound || e.error == DataTransfer::Forbidden) + getFileTransfer()->download(std::move(request), sink); + } 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()); maybeDisable(); throw; @@ -142,12 +142,12 @@ protected: auto callbackPtr = std::make_shared(std::move(callback)); - getDataTransfer()->enqueueDataTransfer(request, - {[callbackPtr, this](std::future result) { + getFileTransfer()->enqueueFileTransfer(request, + {[callbackPtr, this](std::future result) { try { (*callbackPtr)(result.get().data); - } catch (DataTransferError & e) { - if (e.error == DataTransfer::NotFound || e.error == DataTransfer::Forbidden) + } catch (FileTransferError & e) { + if (e.error == FileTransfer::NotFound || e.error == FileTransfer::Forbidden) return (*callbackPtr)(std::shared_ptr()); maybeDisable(); callbackPtr->rethrow(); diff --git a/src/libstore/s3-binary-cache-store.cc b/src/libstore/s3-binary-cache-store.cc index 478cb9f84..fccf010a4 100644 --- a/src/libstore/s3-binary-cache-store.cc +++ b/src/libstore/s3-binary-cache-store.cc @@ -132,7 +132,7 @@ ref S3Helper::makeConfig(const string & region return res; } -S3Helper::DataTransferResult S3Helper::getObject( +S3Helper::FileTransferResult S3Helper::getObject( const std::string & bucketName, const std::string & key) { debug("fetching 's3://%s/%s'...", bucketName, key); @@ -146,7 +146,7 @@ S3Helper::DataTransferResult S3Helper::getObject( return Aws::New("STRINGSTREAM"); }); - DataTransferResult res; + FileTransferResult res; auto now1 = std::chrono::steady_clock::now(); diff --git a/src/libstore/s3.hh b/src/libstore/s3.hh index d7d309243..2042bffcf 100644 --- a/src/libstore/s3.hh +++ b/src/libstore/s3.hh @@ -18,13 +18,13 @@ struct S3Helper ref makeConfig(const std::string & region, const std::string & scheme, const std::string & endpoint); - struct DataTransferResult + struct FileTransferResult { std::shared_ptr data; unsigned int durationMs; }; - DataTransferResult getObject( + FileTransferResult getObject( const std::string & bucketName, const std::string & key); }; diff --git a/src/libutil/logging.cc b/src/libutil/logging.cc index 001a3fb28..bb437cf1c 100644 --- a/src/libutil/logging.cc +++ b/src/libutil/logging.cc @@ -198,7 +198,7 @@ bool handleJSONLogMessage(const std::string & msg, if (action == "start") { auto type = (ActivityType) json["type"]; - if (trusted || type == actDataTransfer) + if (trusted || type == actFileTransfer) activities.emplace(std::piecewise_construct, std::forward_as_tuple(json["id"]), std::forward_as_tuple(*logger, (Verbosity) json["level"], type, diff --git a/src/libutil/logging.hh b/src/libutil/logging.hh index 423ed95da..108b5dcb0 100644 --- a/src/libutil/logging.hh +++ b/src/libutil/logging.hh @@ -17,7 +17,7 @@ typedef enum { typedef enum { actUnknown = 0, actCopyPath = 100, - actDataTransfer = 101, + actFileTransfer = 101, actRealise = 102, actCopyPaths = 103, actBuilds = 104, diff --git a/src/nix-prefetch-url/nix-prefetch-url.cc b/src/nix-prefetch-url/nix-prefetch-url.cc index 4c81e7d82..e136bb95a 100644 --- a/src/nix-prefetch-url/nix-prefetch-url.cc +++ b/src/nix-prefetch-url/nix-prefetch-url.cc @@ -180,9 +180,9 @@ static int _main(int argc, char * * argv) FdSink sink(fd.get()); - DataTransferRequest req(actualUri); + FileTransferRequest req(actualUri); req.decompress = false; - getDataTransfer()->download(std::move(req), sink); + getFileTransfer()->download(std::move(req), sink); } /* Optionally unpack the file. */ diff --git a/src/nix/main.cc b/src/nix/main.cc index 6bb04d8b6..140c963e0 100644 --- a/src/nix/main.cc +++ b/src/nix/main.cc @@ -176,10 +176,10 @@ void mainWrapped(int argc, char * * argv) settings.useSubstitutes = false; if (!settings.tarballTtl.overriden) settings.tarballTtl = std::numeric_limits::max(); - if (!dataTransferSettings.tries.overriden) - dataTransferSettings.tries = 0; - if (!dataTransferSettings.connectTimeout.overriden) - dataTransferSettings.connectTimeout = 1; + if (!fileTransferSettings.tries.overriden) + fileTransferSettings.tries = 0; + if (!fileTransferSettings.connectTimeout.overriden) + fileTransferSettings.connectTimeout = 1; } if (args.refresh) diff --git a/src/nix/progress-bar.cc b/src/nix/progress-bar.cc index d108896ad..adc9b9a5d 100644 --- a/src/nix/progress-bar.cc +++ b/src/nix/progress-bar.cc @@ -190,8 +190,8 @@ public: i->s = fmt("querying " ANSI_BOLD "%s" ANSI_NORMAL " on %s", name, getS(fields, 1)); } - if ((type == actDataTransfer && hasAncestor(*state, actCopyPath, parent)) - || (type == actDataTransfer && hasAncestor(*state, actQueryPathInfo, parent)) + if ((type == actFileTransfer && hasAncestor(*state, actCopyPath, parent)) + || (type == actFileTransfer && hasAncestor(*state, actQueryPathInfo, parent)) || (type == actCopyPath && hasAncestor(*state, actSubstitute, parent))) i->visible = false; @@ -416,7 +416,7 @@ public: if (!s2.empty()) { res += " ("; res += s2; res += ')'; } } - showActivity(actDataTransfer, "%s MiB DL", "%.1f", MiB); + showActivity(actFileTransfer, "%s MiB DL", "%.1f", MiB); { auto s = renderActivity(actOptimiseStore, "%s paths optimised"); diff --git a/src/nix/upgrade-nix.cc b/src/nix/upgrade-nix.cc index a4eda90b4..574f7d8c2 100644 --- a/src/nix/upgrade-nix.cc +++ b/src/nix/upgrade-nix.cc @@ -138,8 +138,8 @@ struct CmdUpgradeNix : MixDryRun, StoreCommand Activity act(*logger, lvlInfo, actUnknown, "querying latest Nix version"); // FIXME: use nixos.org? - auto req = DataTransferRequest(storePathsUrl); - auto res = getDataTransfer()->download(req); + auto req = FileTransferRequest(storePathsUrl); + auto res = getFileTransfer()->download(req); auto state = std::make_unique(Strings(), store); auto v = state->allocValue(); -- cgit v1.2.3 From f5095594e723d02c9bf22fe24ad838ebff4f9165 Mon Sep 17 00:00:00 2001 From: Nikola Knezevic Date: Mon, 6 Apr 2020 23:57:28 +0200 Subject: datatransfer.{cc,hh} -> filetransfer.{cc,hh} --- src/libexpr/common-eval-args.cc | 2 +- src/libexpr/eval.cc | 2 +- src/libexpr/parser.y | 2 +- src/libstore/build.cc | 2 +- src/libstore/builtins/fetchurl.cc | 2 +- src/libstore/datatransfer.cc | 820 ------------------------------- src/libstore/datatransfer.hh | 116 ----- src/libstore/filetransfer.cc | 820 +++++++++++++++++++++++++++++++ src/libstore/filetransfer.hh | 116 +++++ src/libstore/http-binary-cache-store.cc | 2 +- src/libstore/s3-binary-cache-store.cc | 2 +- src/nix-channel/nix-channel.cc | 2 +- src/nix-prefetch-url/nix-prefetch-url.cc | 2 +- src/nix/main.cc | 2 +- src/nix/upgrade-nix.cc | 2 +- 15 files changed, 947 insertions(+), 947 deletions(-) delete mode 100644 src/libstore/datatransfer.cc delete mode 100644 src/libstore/datatransfer.hh create mode 100644 src/libstore/filetransfer.cc create mode 100644 src/libstore/filetransfer.hh (limited to 'src') diff --git a/src/libexpr/common-eval-args.cc b/src/libexpr/common-eval-args.cc index 26460601d..10c9c16bb 100644 --- a/src/libexpr/common-eval-args.cc +++ b/src/libexpr/common-eval-args.cc @@ -1,6 +1,6 @@ #include "common-eval-args.hh" #include "shared.hh" -#include "datatransfer.hh" +#include "filetransfer.hh" #include "util.hh" #include "eval.hh" #include "fetchers.hh" diff --git a/src/libexpr/eval.cc b/src/libexpr/eval.cc index 7a20c6fa6..b91a021b4 100644 --- a/src/libexpr/eval.cc +++ b/src/libexpr/eval.cc @@ -5,7 +5,7 @@ #include "derivations.hh" #include "globals.hh" #include "eval-inline.hh" -#include "datatransfer.hh" +#include "filetransfer.hh" #include "json.hh" #include "function-trace.hh" diff --git a/src/libexpr/parser.y b/src/libexpr/parser.y index 8fbe59301..235ed5b1d 100644 --- a/src/libexpr/parser.y +++ b/src/libexpr/parser.y @@ -544,7 +544,7 @@ formal #include #include "eval.hh" -#include "datatransfer.hh" +#include "filetransfer.hh" #include "fetchers.hh" #include "store-api.hh" diff --git a/src/libstore/build.cc b/src/libstore/build.cc index 9baa2cf1a..0febb8dfb 100644 --- a/src/libstore/build.cc +++ b/src/libstore/build.cc @@ -7,7 +7,7 @@ #include "affinity.hh" #include "builtins.hh" #include "builtins/buildenv.hh" -#include "datatransfer.hh" +#include "filetransfer.hh" #include "finally.hh" #include "compression.hh" #include "json.hh" diff --git a/src/libstore/builtins/fetchurl.cc b/src/libstore/builtins/fetchurl.cc index 1432de9d7..486babf14 100644 --- a/src/libstore/builtins/fetchurl.cc +++ b/src/libstore/builtins/fetchurl.cc @@ -1,5 +1,5 @@ #include "builtins.hh" -#include "datatransfer.hh" +#include "filetransfer.hh" #include "store-api.hh" #include "archive.hh" #include "compression.hh" diff --git a/src/libstore/datatransfer.cc b/src/libstore/datatransfer.cc deleted file mode 100644 index 5cc63a1d8..000000000 --- a/src/libstore/datatransfer.cc +++ /dev/null @@ -1,820 +0,0 @@ -#include "datatransfer.hh" -#include "util.hh" -#include "globals.hh" -#include "store-api.hh" -#include "s3.hh" -#include "compression.hh" -#include "finally.hh" - -#ifdef ENABLE_S3 -#include -#endif - -#include -#include - -#include - -#include -#include -#include -#include -#include -#include -#include - -using namespace std::string_literals; - -namespace nix { - -FileTransferSettings fileTransferSettings; - -static GlobalConfig::Register r1(&fileTransferSettings); - -std::string resolveUri(const std::string & uri) -{ - if (uri.compare(0, 8, "channel:") == 0) - return "https://nixos.org/channels/" + std::string(uri, 8) + "/nixexprs.tar.xz"; - else - return uri; -} - -struct curlFileTransfer : public FileTransfer -{ - CURLM * curlm = 0; - - std::random_device rd; - std::mt19937 mt19937; - - struct TransferItem : public std::enable_shared_from_this - { - curlFileTransfer & fileTransfer; - FileTransferRequest request; - FileTransferResult result; - Activity act; - bool done = false; // whether either the success or failure function has been called - Callback callback; - CURL * req = 0; - bool active = false; // whether the handle has been added to the multi object - std::string status; - - unsigned int attempt = 0; - - /* Don't start this download until the specified time point - has been reached. */ - std::chrono::steady_clock::time_point embargo; - - struct curl_slist * requestHeaders = 0; - - std::string encoding; - - bool acceptRanges = false; - - curl_off_t writtenToSink = 0; - - TransferItem(curlFileTransfer & fileTransfer, - const FileTransferRequest & request, - Callback && callback) - : fileTransfer(fileTransfer) - , request(request) - , act(*logger, lvlTalkative, actFileTransfer, - fmt(request.data ? "uploading '%s'" : "downloading '%s'", request.uri), - {request.uri}, request.parentAct) - , callback(std::move(callback)) - , finalSink([this](const unsigned char * data, size_t len) { - if (this->request.dataCallback) { - long httpStatus = 0; - curl_easy_getinfo(req, CURLINFO_RESPONSE_CODE, &httpStatus); - - /* Only write data to the sink if this is a - successful response. */ - if (httpStatus == 0 || httpStatus == 200 || httpStatus == 201 || httpStatus == 206) { - writtenToSink += len; - this->request.dataCallback((char *) data, len); - } - } else - this->result.data->append((char *) data, len); - }) - { - if (!request.expectedETag.empty()) - requestHeaders = curl_slist_append(requestHeaders, ("If-None-Match: " + request.expectedETag).c_str()); - if (!request.mimeType.empty()) - requestHeaders = curl_slist_append(requestHeaders, ("Content-Type: " + request.mimeType).c_str()); - } - - ~TransferItem() - { - if (req) { - if (active) - curl_multi_remove_handle(fileTransfer.curlm, req); - curl_easy_cleanup(req); - } - if (requestHeaders) curl_slist_free_all(requestHeaders); - try { - if (!done) - fail(FileTransferError(Interrupted, format("download of '%s' was interrupted") % request.uri)); - } catch (...) { - ignoreException(); - } - } - - void failEx(std::exception_ptr ex) - { - assert(!done); - done = true; - callback.rethrow(ex); - } - - template - void fail(const T & e) - { - failEx(std::make_exception_ptr(e)); - } - - LambdaSink finalSink; - std::shared_ptr decompressionSink; - - std::exception_ptr writeException; - - size_t writeCallback(void * contents, size_t size, size_t nmemb) - { - try { - size_t realSize = size * nmemb; - result.bodySize += realSize; - - if (!decompressionSink) - decompressionSink = makeDecompressionSink(encoding, finalSink); - - (*decompressionSink)((unsigned char *) contents, realSize); - - return realSize; - } catch (...) { - writeException = std::current_exception(); - return 0; - } - } - - static size_t writeCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) - { - return ((TransferItem *) userp)->writeCallback(contents, size, nmemb); - } - - size_t headerCallback(void * contents, size_t size, size_t nmemb) - { - size_t realSize = size * nmemb; - std::string line((char *) contents, realSize); - printMsg(lvlVomit, format("got header for '%s': %s") % request.uri % trim(line)); - if (line.compare(0, 5, "HTTP/") == 0) { // new response starts - result.etag = ""; - auto ss = tokenizeString>(line, " "); - status = ss.size() >= 2 ? ss[1] : ""; - result.data = std::make_shared(); - result.bodySize = 0; - acceptRanges = false; - encoding = ""; - } else { - auto i = line.find(':'); - if (i != string::npos) { - string name = toLower(trim(string(line, 0, i))); - if (name == "etag") { - result.etag = trim(string(line, i + 1)); - /* Hack to work around a GitHub bug: it sends - ETags, but ignores If-None-Match. So if we get - the expected ETag on a 200 response, then shut - down the connection because we already have the - data. */ - if (result.etag == request.expectedETag && status == "200") { - debug(format("shutting down on 200 HTTP response with expected ETag")); - return 0; - } - } else if (name == "content-encoding") - encoding = trim(string(line, i + 1)); - else if (name == "accept-ranges" && toLower(trim(std::string(line, i + 1))) == "bytes") - acceptRanges = true; - } - } - return realSize; - } - - static size_t headerCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) - { - return ((TransferItem *) userp)->headerCallback(contents, size, nmemb); - } - - int progressCallback(double dltotal, double dlnow) - { - try { - act.progress(dlnow, dltotal); - } catch (nix::Interrupted &) { - assert(_isInterrupted); - } - return _isInterrupted; - } - - static int progressCallbackWrapper(void * userp, double dltotal, double dlnow, double ultotal, double ulnow) - { - return ((TransferItem *) userp)->progressCallback(dltotal, dlnow); - } - - static int debugCallback(CURL * handle, curl_infotype type, char * data, size_t size, void * userptr) - { - if (type == CURLINFO_TEXT) - vomit("curl: %s", chomp(std::string(data, size))); - return 0; - } - - size_t readOffset = 0; - size_t readCallback(char *buffer, size_t size, size_t nitems) - { - if (readOffset == request.data->length()) - return 0; - auto count = std::min(size * nitems, request.data->length() - readOffset); - assert(count); - memcpy(buffer, request.data->data() + readOffset, count); - readOffset += count; - return count; - } - - static size_t readCallbackWrapper(char *buffer, size_t size, size_t nitems, void * userp) - { - return ((TransferItem *) userp)->readCallback(buffer, size, nitems); - } - - void init() - { - if (!req) req = curl_easy_init(); - - curl_easy_reset(req); - - if (verbosity >= lvlVomit) { - curl_easy_setopt(req, CURLOPT_VERBOSE, 1); - curl_easy_setopt(req, CURLOPT_DEBUGFUNCTION, TransferItem::debugCallback); - } - - curl_easy_setopt(req, CURLOPT_URL, request.uri.c_str()); - curl_easy_setopt(req, CURLOPT_FOLLOWLOCATION, 1L); - curl_easy_setopt(req, CURLOPT_MAXREDIRS, 10); - curl_easy_setopt(req, CURLOPT_NOSIGNAL, 1); - curl_easy_setopt(req, CURLOPT_USERAGENT, - ("curl/" LIBCURL_VERSION " Nix/" + nixVersion + - (fileTransferSettings.userAgentSuffix != "" ? " " + fileTransferSettings.userAgentSuffix.get() : "")).c_str()); - #if LIBCURL_VERSION_NUM >= 0x072b00 - curl_easy_setopt(req, CURLOPT_PIPEWAIT, 1); - #endif - #if LIBCURL_VERSION_NUM >= 0x072f00 - if (fileTransferSettings.enableHttp2) - curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS); - else - curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1); - #endif - curl_easy_setopt(req, CURLOPT_WRITEFUNCTION, TransferItem::writeCallbackWrapper); - curl_easy_setopt(req, CURLOPT_WRITEDATA, this); - curl_easy_setopt(req, CURLOPT_HEADERFUNCTION, TransferItem::headerCallbackWrapper); - curl_easy_setopt(req, CURLOPT_HEADERDATA, this); - - curl_easy_setopt(req, CURLOPT_PROGRESSFUNCTION, progressCallbackWrapper); - curl_easy_setopt(req, CURLOPT_PROGRESSDATA, this); - curl_easy_setopt(req, CURLOPT_NOPROGRESS, 0); - - curl_easy_setopt(req, CURLOPT_HTTPHEADER, requestHeaders); - - if (request.head) - curl_easy_setopt(req, CURLOPT_NOBODY, 1); - - if (request.data) { - curl_easy_setopt(req, CURLOPT_UPLOAD, 1L); - curl_easy_setopt(req, CURLOPT_READFUNCTION, readCallbackWrapper); - curl_easy_setopt(req, CURLOPT_READDATA, this); - curl_easy_setopt(req, CURLOPT_INFILESIZE_LARGE, (curl_off_t) request.data->length()); - } - - if (request.verifyTLS) { - debug("verify TLS: Nix CA file = '%s'", settings.caFile); - if (settings.caFile != "") - curl_easy_setopt(req, CURLOPT_CAINFO, settings.caFile.c_str()); - } else { - curl_easy_setopt(req, CURLOPT_SSL_VERIFYPEER, 0); - curl_easy_setopt(req, CURLOPT_SSL_VERIFYHOST, 0); - } - - curl_easy_setopt(req, CURLOPT_CONNECTTIMEOUT, fileTransferSettings.connectTimeout.get()); - - curl_easy_setopt(req, CURLOPT_LOW_SPEED_LIMIT, 1L); - curl_easy_setopt(req, CURLOPT_LOW_SPEED_TIME, fileTransferSettings.stalledDownloadTimeout.get()); - - /* If no file exist in the specified path, curl continues to work - anyway as if netrc support was disabled. */ - curl_easy_setopt(req, CURLOPT_NETRC_FILE, settings.netrcFile.get().c_str()); - curl_easy_setopt(req, CURLOPT_NETRC, CURL_NETRC_OPTIONAL); - - if (writtenToSink) - curl_easy_setopt(req, CURLOPT_RESUME_FROM_LARGE, writtenToSink); - - result.data = std::make_shared(); - result.bodySize = 0; - } - - void finish(CURLcode code) - { - long httpStatus = 0; - curl_easy_getinfo(req, CURLINFO_RESPONSE_CODE, &httpStatus); - - char * effectiveUriCStr; - curl_easy_getinfo(req, CURLINFO_EFFECTIVE_URL, &effectiveUriCStr); - if (effectiveUriCStr) - result.effectiveUri = effectiveUriCStr; - - debug("finished %s of '%s'; curl status = %d, HTTP status = %d, body = %d bytes", - request.verb(), request.uri, code, httpStatus, result.bodySize); - - if (decompressionSink) { - try { - decompressionSink->finish(); - } catch (...) { - writeException = std::current_exception(); - } - } - - if (code == CURLE_WRITE_ERROR && result.etag == request.expectedETag) { - code = CURLE_OK; - httpStatus = 304; - } - - if (writeException) - failEx(writeException); - - else if (code == CURLE_OK && - (httpStatus == 200 || httpStatus == 201 || httpStatus == 204 || httpStatus == 206 || httpStatus == 304 || httpStatus == 226 /* FTP */ || httpStatus == 0 /* other protocol */)) - { - result.cached = httpStatus == 304; - act.progress(result.bodySize, result.bodySize); - done = true; - callback(std::move(result)); - } - - else { - // We treat most errors as transient, but won't retry when hopeless - Error err = Transient; - - if (httpStatus == 404 || httpStatus == 410 || code == CURLE_FILE_COULDNT_READ_FILE) { - // The file is definitely not there - err = NotFound; - } else if (httpStatus == 401 || httpStatus == 403 || httpStatus == 407) { - // Don't retry on authentication/authorization failures - err = Forbidden; - } else if (httpStatus >= 400 && httpStatus < 500 && httpStatus != 408 && httpStatus != 429) { - // Most 4xx errors are client errors and are probably not worth retrying: - // * 408 means the server timed out waiting for us, so we try again - // * 429 means too many requests, so we retry (with a delay) - err = Misc; - } else if (httpStatus == 501 || httpStatus == 505 || httpStatus == 511) { - // Let's treat most 5xx (server) errors as transient, except for a handful: - // * 501 not implemented - // * 505 http version not supported - // * 511 we're behind a captive portal - err = Misc; - } else { - // Don't bother retrying on certain cURL errors either - switch (code) { - case CURLE_FAILED_INIT: - case CURLE_URL_MALFORMAT: - case CURLE_NOT_BUILT_IN: - case CURLE_REMOTE_ACCESS_DENIED: - case CURLE_FILE_COULDNT_READ_FILE: - case CURLE_FUNCTION_NOT_FOUND: - case CURLE_ABORTED_BY_CALLBACK: - case CURLE_BAD_FUNCTION_ARGUMENT: - case CURLE_INTERFACE_FAILED: - case CURLE_UNKNOWN_OPTION: - case CURLE_SSL_CACERT_BADFILE: - case CURLE_TOO_MANY_REDIRECTS: - case CURLE_WRITE_ERROR: - case CURLE_UNSUPPORTED_PROTOCOL: - err = Misc; - break; - default: // Shut up warnings - break; - } - } - - attempt++; - - auto exc = - code == CURLE_ABORTED_BY_CALLBACK && _isInterrupted - ? FileTransferError(Interrupted, fmt("%s of '%s' was interrupted", request.verb(), request.uri)) - : httpStatus != 0 - ? FileTransferError(err, - fmt("unable to %s '%s': HTTP error %d", - request.verb(), request.uri, httpStatus) - + (code == CURLE_OK ? "" : fmt(" (curl error: %s)", curl_easy_strerror(code))) - ) - : FileTransferError(err, - fmt("unable to %s '%s': %s (%d)", - request.verb(), request.uri, curl_easy_strerror(code), code)); - - /* If this is a transient error, then maybe retry the - download after a while. If we're writing to a - sink, we can only retry if the server supports - ranged requests. */ - if (err == Transient - && attempt < request.tries - && (!this->request.dataCallback - || writtenToSink == 0 - || (acceptRanges && encoding.empty()))) - { - int ms = request.baseRetryTimeMs * std::pow(2.0f, attempt - 1 + std::uniform_real_distribution<>(0.0, 0.5)(fileTransfer.mt19937)); - if (writtenToSink) - warn("%s; retrying from offset %d in %d ms", exc.what(), writtenToSink, ms); - else - warn("%s; retrying in %d ms", exc.what(), ms); - embargo = std::chrono::steady_clock::now() + std::chrono::milliseconds(ms); - fileTransfer.enqueueItem(shared_from_this()); - } - else - fail(exc); - } - } - }; - - struct State - { - struct EmbargoComparator { - bool operator() (const std::shared_ptr & i1, const std::shared_ptr & i2) { - return i1->embargo > i2->embargo; - } - }; - bool quit = false; - std::priority_queue, std::vector>, EmbargoComparator> incoming; - }; - - Sync state_; - - /* We can't use a std::condition_variable to wake up the curl - thread, because it only monitors file descriptors. So use a - pipe instead. */ - Pipe wakeupPipe; - - std::thread workerThread; - - curlFileTransfer() - : mt19937(rd()) - { - static std::once_flag globalInit; - std::call_once(globalInit, curl_global_init, CURL_GLOBAL_ALL); - - curlm = curl_multi_init(); - - #if LIBCURL_VERSION_NUM >= 0x072b00 // Multiplex requires >= 7.43.0 - curl_multi_setopt(curlm, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX); - #endif - #if LIBCURL_VERSION_NUM >= 0x071e00 // Max connections requires >= 7.30.0 - curl_multi_setopt(curlm, CURLMOPT_MAX_TOTAL_CONNECTIONS, - fileTransferSettings.httpConnections.get()); - #endif - - wakeupPipe.create(); - fcntl(wakeupPipe.readSide.get(), F_SETFL, O_NONBLOCK); - - workerThread = std::thread([&]() { workerThreadEntry(); }); - } - - ~curlFileTransfer() - { - stopWorkerThread(); - - workerThread.join(); - - if (curlm) curl_multi_cleanup(curlm); - } - - void stopWorkerThread() - { - /* Signal the worker thread to exit. */ - { - auto state(state_.lock()); - state->quit = true; - } - writeFull(wakeupPipe.writeSide.get(), " ", false); - } - - void workerThreadMain() - { - /* Cause this thread to be notified on SIGINT. */ - auto callback = createInterruptCallback([&]() { - stopWorkerThread(); - }); - - std::map> items; - - bool quit = false; - - std::chrono::steady_clock::time_point nextWakeup; - - while (!quit) { - checkInterrupt(); - - /* Let curl do its thing. */ - int running; - CURLMcode mc = curl_multi_perform(curlm, &running); - if (mc != CURLM_OK) - throw nix::Error(format("unexpected error from curl_multi_perform(): %s") % curl_multi_strerror(mc)); - - /* Set the promises of any finished requests. */ - CURLMsg * msg; - int left; - while ((msg = curl_multi_info_read(curlm, &left))) { - if (msg->msg == CURLMSG_DONE) { - auto i = items.find(msg->easy_handle); - assert(i != items.end()); - i->second->finish(msg->data.result); - curl_multi_remove_handle(curlm, i->second->req); - i->second->active = false; - items.erase(i); - } - } - - /* Wait for activity, including wakeup events. */ - int numfds = 0; - struct curl_waitfd extraFDs[1]; - extraFDs[0].fd = wakeupPipe.readSide.get(); - extraFDs[0].events = CURL_WAIT_POLLIN; - extraFDs[0].revents = 0; - long maxSleepTimeMs = items.empty() ? 10000 : 100; - auto sleepTimeMs = - nextWakeup != std::chrono::steady_clock::time_point() - ? std::max(0, (int) std::chrono::duration_cast(nextWakeup - std::chrono::steady_clock::now()).count()) - : maxSleepTimeMs; - vomit("download thread waiting for %d ms", sleepTimeMs); - mc = curl_multi_wait(curlm, extraFDs, 1, sleepTimeMs, &numfds); - if (mc != CURLM_OK) - throw nix::Error(format("unexpected error from curl_multi_wait(): %s") % curl_multi_strerror(mc)); - - nextWakeup = std::chrono::steady_clock::time_point(); - - /* Add new curl requests from the incoming requests queue, - except for requests that are embargoed (waiting for a - retry timeout to expire). */ - if (extraFDs[0].revents & CURL_WAIT_POLLIN) { - char buf[1024]; - auto res = read(extraFDs[0].fd, buf, sizeof(buf)); - if (res == -1 && errno != EINTR) - throw SysError("reading curl wakeup socket"); - } - - std::vector> incoming; - auto now = std::chrono::steady_clock::now(); - - { - auto state(state_.lock()); - while (!state->incoming.empty()) { - auto item = state->incoming.top(); - if (item->embargo <= now) { - incoming.push_back(item); - state->incoming.pop(); - } else { - if (nextWakeup == std::chrono::steady_clock::time_point() - || item->embargo < nextWakeup) - nextWakeup = item->embargo; - break; - } - } - quit = state->quit; - } - - for (auto & item : incoming) { - debug("starting %s of %s", item->request.verb(), item->request.uri); - item->init(); - curl_multi_add_handle(curlm, item->req); - item->active = true; - items[item->req] = item; - } - } - - debug("download thread shutting down"); - } - - void workerThreadEntry() - { - try { - workerThreadMain(); - } catch (nix::Interrupted & e) { - } catch (std::exception & e) { - printError("unexpected error in download thread: %s", e.what()); - } - - { - auto state(state_.lock()); - while (!state->incoming.empty()) state->incoming.pop(); - state->quit = true; - } - } - - void enqueueItem(std::shared_ptr item) - { - if (item->request.data - && !hasPrefix(item->request.uri, "http://") - && !hasPrefix(item->request.uri, "https://")) - throw nix::Error("uploading to '%s' is not supported", item->request.uri); - - { - auto state(state_.lock()); - if (state->quit) - throw nix::Error("cannot enqueue download request because the download thread is shutting down"); - state->incoming.push(item); - } - writeFull(wakeupPipe.writeSide.get(), " "); - } - -#ifdef ENABLE_S3 - std::tuple parseS3Uri(std::string uri) - { - auto [path, params] = splitUriAndParams(uri); - - auto slash = path.find('/', 5); // 5 is the length of "s3://" prefix - if (slash == std::string::npos) - throw nix::Error("bad S3 URI '%s'", path); - - std::string bucketName(path, 5, slash - 5); - std::string key(path, slash + 1); - - return {bucketName, key, params}; - } -#endif - - void enqueueFileTransfer(const FileTransferRequest & request, - Callback callback) override - { - /* Ugly hack to support s3:// URIs. */ - if (hasPrefix(request.uri, "s3://")) { - // FIXME: do this on a worker thread - try { -#ifdef ENABLE_S3 - auto [bucketName, key, params] = parseS3Uri(request.uri); - - std::string profile = get(params, "profile").value_or(""); - std::string region = get(params, "region").value_or(Aws::Region::US_EAST_1); - std::string scheme = get(params, "scheme").value_or(""); - std::string endpoint = get(params, "endpoint").value_or(""); - - S3Helper s3Helper(profile, region, scheme, endpoint); - - // FIXME: implement ETag - auto s3Res = s3Helper.getObject(bucketName, key); - FileTransferResult res; - if (!s3Res.data) - throw FileTransferError(NotFound, fmt("S3 object '%s' does not exist", request.uri)); - res.data = s3Res.data; - callback(std::move(res)); -#else - throw nix::Error("cannot download '%s' because Nix is not built with S3 support", request.uri); -#endif - } catch (...) { callback.rethrow(); } - return; - } - - enqueueItem(std::make_shared(*this, request, std::move(callback))); - } -}; - -ref getFileTransfer() -{ - static ref fileTransfer = makeFileTransfer(); - return fileTransfer; -} - -ref makeFileTransfer() -{ - return make_ref(); -} - -std::future FileTransfer::enqueueFileTransfer(const FileTransferRequest & request) -{ - auto promise = std::make_shared>(); - enqueueFileTransfer(request, - {[promise](std::future fut) { - try { - promise->set_value(fut.get()); - } catch (...) { - promise->set_exception(std::current_exception()); - } - }}); - return promise->get_future(); -} - -FileTransferResult FileTransfer::download(const FileTransferRequest & request) -{ - return enqueueFileTransfer(request).get(); -} - -FileTransferResult FileTransfer::upload(const FileTransferRequest & request) -{ - /* Note: this method is the same as download, but helps in readability */ - return enqueueFileTransfer(request).get(); -} - -void FileTransfer::download(FileTransferRequest && request, Sink & sink) -{ - /* Note: we can't call 'sink' via request.dataCallback, because - that would cause the sink to execute on the fileTransfer - thread. If 'sink' is a coroutine, this will fail. Also, if the - sink is expensive (e.g. one that does decompression and writing - to the Nix store), it would stall the download thread too much. - Therefore we use a buffer to communicate data between the - download thread and the calling thread. */ - - struct State { - bool quit = false; - std::exception_ptr exc; - std::string data; - std::condition_variable avail, request; - }; - - auto _state = std::make_shared>(); - - /* In case of an exception, wake up the download thread. FIXME: - abort the download request. */ - Finally finally([&]() { - auto state(_state->lock()); - state->quit = true; - state->request.notify_one(); - }); - - request.dataCallback = [_state](char * buf, size_t len) { - - auto state(_state->lock()); - - if (state->quit) return; - - /* If the buffer is full, then go to sleep until the calling - thread wakes us up (i.e. when it has removed data from the - buffer). We don't wait forever to prevent stalling the - download thread. (Hopefully sleeping will throttle the - sender.) */ - if (state->data.size() > 1024 * 1024) { - debug("download buffer is full; going to sleep"); - state.wait_for(state->request, std::chrono::seconds(10)); - } - - /* Append data to the buffer and wake up the calling - thread. */ - state->data.append(buf, len); - state->avail.notify_one(); - }; - - enqueueFileTransfer(request, - {[_state](std::future fut) { - auto state(_state->lock()); - state->quit = true; - try { - fut.get(); - } catch (...) { - state->exc = std::current_exception(); - } - state->avail.notify_one(); - state->request.notify_one(); - }}); - - while (true) { - checkInterrupt(); - - std::string chunk; - - /* Grab data if available, otherwise wait for the download - thread to wake us up. */ - { - auto state(_state->lock()); - - while (state->data.empty()) { - - if (state->quit) { - if (state->exc) std::rethrow_exception(state->exc); - return; - } - - state.wait(state->avail); - } - - chunk = std::move(state->data); - - state->request.notify_one(); - } - - /* Flush the data to the sink and wake up the download thread - if it's blocked on a full buffer. We don't hold the state - lock while doing this to prevent blocking the download - thread if sink() takes a long time. */ - sink((unsigned char *) chunk.data(), chunk.size()); - } -} - -bool isUri(const string & s) -{ - if (s.compare(0, 8, "channel:") == 0) return true; - size_t pos = s.find("://"); - if (pos == string::npos) return false; - string scheme(s, 0, pos); - return scheme == "http" || scheme == "https" || scheme == "file" || scheme == "channel" || scheme == "git" || scheme == "s3" || scheme == "ssh"; -} - - -} diff --git a/src/libstore/datatransfer.hh b/src/libstore/datatransfer.hh deleted file mode 100644 index 2347f363d..000000000 --- a/src/libstore/datatransfer.hh +++ /dev/null @@ -1,116 +0,0 @@ -#pragma once - -#include "types.hh" -#include "hash.hh" -#include "config.hh" - -#include -#include - -namespace nix { - -struct FileTransferSettings : Config -{ - Setting enableHttp2{this, true, "http2", - "Whether to enable HTTP/2 support."}; - - Setting userAgentSuffix{this, "", "user-agent-suffix", - "String appended to the user agent in HTTP requests."}; - - Setting httpConnections{this, 25, "http-connections", - "Number of parallel HTTP connections.", - {"binary-caches-parallel-connections"}}; - - Setting connectTimeout{this, 0, "connect-timeout", - "Timeout for connecting to servers during downloads. 0 means use curl's builtin default."}; - - Setting stalledDownloadTimeout{this, 300, "stalled-download-timeout", - "Timeout (in seconds) for receiving data from servers during download. Nix cancels idle downloads after this timeout's duration."}; - - Setting tries{this, 5, "download-attempts", - "How often Nix will attempt to download a file before giving up."}; -}; - -extern FileTransferSettings fileTransferSettings; - -struct FileTransferRequest -{ - std::string uri; - std::string expectedETag; - bool verifyTLS = true; - bool head = false; - size_t tries = fileTransferSettings.tries; - unsigned int baseRetryTimeMs = 250; - ActivityId parentAct; - bool decompress = true; - std::shared_ptr data; - std::string mimeType; - std::function dataCallback; - - FileTransferRequest(const std::string & uri) - : uri(uri), parentAct(getCurActivity()) { } - - std::string verb() - { - return data ? "upload" : "download"; - } -}; - -struct FileTransferResult -{ - bool cached = false; - std::string etag; - std::string effectiveUri; - std::shared_ptr data; - uint64_t bodySize = 0; -}; - -class Store; - -struct FileTransfer -{ - virtual ~FileTransfer() { } - - /* Enqueue a data transfer request, returning a future to the result of - the download. The future may throw a FileTransferError - exception. */ - virtual void enqueueFileTransfer(const FileTransferRequest & request, - Callback callback) = 0; - - std::future enqueueFileTransfer(const FileTransferRequest & request); - - /* Synchronously download a file. */ - FileTransferResult download(const FileTransferRequest & request); - - /* Synchronously upload a file. */ - FileTransferResult upload(const FileTransferRequest & request); - - /* Download a file, writing its data to a sink. The sink will be - invoked on the thread of the caller. */ - void download(FileTransferRequest && request, Sink & sink); - - enum Error { NotFound, Forbidden, Misc, Transient, Interrupted }; -}; - -/* Return a shared FileTransfer object. Using this object is preferred - because it enables connection reuse and HTTP/2 multiplexing. */ -ref getFileTransfer(); - -/* Return a new FileTransfer object. */ -ref makeFileTransfer(); - -class FileTransferError : public Error -{ -public: - FileTransfer::Error error; - FileTransferError(FileTransfer::Error error, const FormatOrString & fs) - : Error(fs), error(error) - { } -}; - -bool isUri(const string & s); - -/* Resolve deprecated 'channel:' URLs. */ -std::string resolveUri(const std::string & uri); - -} diff --git a/src/libstore/filetransfer.cc b/src/libstore/filetransfer.cc new file mode 100644 index 000000000..e9684b3d4 --- /dev/null +++ b/src/libstore/filetransfer.cc @@ -0,0 +1,820 @@ +#include "filetransfer.hh" +#include "util.hh" +#include "globals.hh" +#include "store-api.hh" +#include "s3.hh" +#include "compression.hh" +#include "finally.hh" + +#ifdef ENABLE_S3 +#include +#endif + +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include + +using namespace std::string_literals; + +namespace nix { + +FileTransferSettings fileTransferSettings; + +static GlobalConfig::Register r1(&fileTransferSettings); + +std::string resolveUri(const std::string & uri) +{ + if (uri.compare(0, 8, "channel:") == 0) + return "https://nixos.org/channels/" + std::string(uri, 8) + "/nixexprs.tar.xz"; + else + return uri; +} + +struct curlFileTransfer : public FileTransfer +{ + CURLM * curlm = 0; + + std::random_device rd; + std::mt19937 mt19937; + + struct TransferItem : public std::enable_shared_from_this + { + curlFileTransfer & fileTransfer; + FileTransferRequest request; + FileTransferResult result; + Activity act; + bool done = false; // whether either the success or failure function has been called + Callback callback; + CURL * req = 0; + bool active = false; // whether the handle has been added to the multi object + std::string status; + + unsigned int attempt = 0; + + /* Don't start this download until the specified time point + has been reached. */ + std::chrono::steady_clock::time_point embargo; + + struct curl_slist * requestHeaders = 0; + + std::string encoding; + + bool acceptRanges = false; + + curl_off_t writtenToSink = 0; + + TransferItem(curlFileTransfer & fileTransfer, + const FileTransferRequest & request, + Callback && callback) + : fileTransfer(fileTransfer) + , request(request) + , act(*logger, lvlTalkative, actFileTransfer, + fmt(request.data ? "uploading '%s'" : "downloading '%s'", request.uri), + {request.uri}, request.parentAct) + , callback(std::move(callback)) + , finalSink([this](const unsigned char * data, size_t len) { + if (this->request.dataCallback) { + long httpStatus = 0; + curl_easy_getinfo(req, CURLINFO_RESPONSE_CODE, &httpStatus); + + /* Only write data to the sink if this is a + successful response. */ + if (httpStatus == 0 || httpStatus == 200 || httpStatus == 201 || httpStatus == 206) { + writtenToSink += len; + this->request.dataCallback((char *) data, len); + } + } else + this->result.data->append((char *) data, len); + }) + { + if (!request.expectedETag.empty()) + requestHeaders = curl_slist_append(requestHeaders, ("If-None-Match: " + request.expectedETag).c_str()); + if (!request.mimeType.empty()) + requestHeaders = curl_slist_append(requestHeaders, ("Content-Type: " + request.mimeType).c_str()); + } + + ~TransferItem() + { + if (req) { + if (active) + curl_multi_remove_handle(fileTransfer.curlm, req); + curl_easy_cleanup(req); + } + if (requestHeaders) curl_slist_free_all(requestHeaders); + try { + if (!done) + fail(FileTransferError(Interrupted, format("download of '%s' was interrupted") % request.uri)); + } catch (...) { + ignoreException(); + } + } + + void failEx(std::exception_ptr ex) + { + assert(!done); + done = true; + callback.rethrow(ex); + } + + template + void fail(const T & e) + { + failEx(std::make_exception_ptr(e)); + } + + LambdaSink finalSink; + std::shared_ptr decompressionSink; + + std::exception_ptr writeException; + + size_t writeCallback(void * contents, size_t size, size_t nmemb) + { + try { + size_t realSize = size * nmemb; + result.bodySize += realSize; + + if (!decompressionSink) + decompressionSink = makeDecompressionSink(encoding, finalSink); + + (*decompressionSink)((unsigned char *) contents, realSize); + + return realSize; + } catch (...) { + writeException = std::current_exception(); + return 0; + } + } + + static size_t writeCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) + { + return ((TransferItem *) userp)->writeCallback(contents, size, nmemb); + } + + size_t headerCallback(void * contents, size_t size, size_t nmemb) + { + size_t realSize = size * nmemb; + std::string line((char *) contents, realSize); + printMsg(lvlVomit, format("got header for '%s': %s") % request.uri % trim(line)); + if (line.compare(0, 5, "HTTP/") == 0) { // new response starts + result.etag = ""; + auto ss = tokenizeString>(line, " "); + status = ss.size() >= 2 ? ss[1] : ""; + result.data = std::make_shared(); + result.bodySize = 0; + acceptRanges = false; + encoding = ""; + } else { + auto i = line.find(':'); + if (i != string::npos) { + string name = toLower(trim(string(line, 0, i))); + if (name == "etag") { + result.etag = trim(string(line, i + 1)); + /* Hack to work around a GitHub bug: it sends + ETags, but ignores If-None-Match. So if we get + the expected ETag on a 200 response, then shut + down the connection because we already have the + data. */ + if (result.etag == request.expectedETag && status == "200") { + debug(format("shutting down on 200 HTTP response with expected ETag")); + return 0; + } + } else if (name == "content-encoding") + encoding = trim(string(line, i + 1)); + else if (name == "accept-ranges" && toLower(trim(std::string(line, i + 1))) == "bytes") + acceptRanges = true; + } + } + return realSize; + } + + static size_t headerCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) + { + return ((TransferItem *) userp)->headerCallback(contents, size, nmemb); + } + + int progressCallback(double dltotal, double dlnow) + { + try { + act.progress(dlnow, dltotal); + } catch (nix::Interrupted &) { + assert(_isInterrupted); + } + return _isInterrupted; + } + + static int progressCallbackWrapper(void * userp, double dltotal, double dlnow, double ultotal, double ulnow) + { + return ((TransferItem *) userp)->progressCallback(dltotal, dlnow); + } + + static int debugCallback(CURL * handle, curl_infotype type, char * data, size_t size, void * userptr) + { + if (type == CURLINFO_TEXT) + vomit("curl: %s", chomp(std::string(data, size))); + return 0; + } + + size_t readOffset = 0; + size_t readCallback(char *buffer, size_t size, size_t nitems) + { + if (readOffset == request.data->length()) + return 0; + auto count = std::min(size * nitems, request.data->length() - readOffset); + assert(count); + memcpy(buffer, request.data->data() + readOffset, count); + readOffset += count; + return count; + } + + static size_t readCallbackWrapper(char *buffer, size_t size, size_t nitems, void * userp) + { + return ((TransferItem *) userp)->readCallback(buffer, size, nitems); + } + + void init() + { + if (!req) req = curl_easy_init(); + + curl_easy_reset(req); + + if (verbosity >= lvlVomit) { + curl_easy_setopt(req, CURLOPT_VERBOSE, 1); + curl_easy_setopt(req, CURLOPT_DEBUGFUNCTION, TransferItem::debugCallback); + } + + curl_easy_setopt(req, CURLOPT_URL, request.uri.c_str()); + curl_easy_setopt(req, CURLOPT_FOLLOWLOCATION, 1L); + curl_easy_setopt(req, CURLOPT_MAXREDIRS, 10); + curl_easy_setopt(req, CURLOPT_NOSIGNAL, 1); + curl_easy_setopt(req, CURLOPT_USERAGENT, + ("curl/" LIBCURL_VERSION " Nix/" + nixVersion + + (fileTransferSettings.userAgentSuffix != "" ? " " + fileTransferSettings.userAgentSuffix.get() : "")).c_str()); + #if LIBCURL_VERSION_NUM >= 0x072b00 + curl_easy_setopt(req, CURLOPT_PIPEWAIT, 1); + #endif + #if LIBCURL_VERSION_NUM >= 0x072f00 + if (fileTransferSettings.enableHttp2) + curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS); + else + curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1); + #endif + curl_easy_setopt(req, CURLOPT_WRITEFUNCTION, TransferItem::writeCallbackWrapper); + curl_easy_setopt(req, CURLOPT_WRITEDATA, this); + curl_easy_setopt(req, CURLOPT_HEADERFUNCTION, TransferItem::headerCallbackWrapper); + curl_easy_setopt(req, CURLOPT_HEADERDATA, this); + + curl_easy_setopt(req, CURLOPT_PROGRESSFUNCTION, progressCallbackWrapper); + curl_easy_setopt(req, CURLOPT_PROGRESSDATA, this); + curl_easy_setopt(req, CURLOPT_NOPROGRESS, 0); + + curl_easy_setopt(req, CURLOPT_HTTPHEADER, requestHeaders); + + if (request.head) + curl_easy_setopt(req, CURLOPT_NOBODY, 1); + + if (request.data) { + curl_easy_setopt(req, CURLOPT_UPLOAD, 1L); + curl_easy_setopt(req, CURLOPT_READFUNCTION, readCallbackWrapper); + curl_easy_setopt(req, CURLOPT_READDATA, this); + curl_easy_setopt(req, CURLOPT_INFILESIZE_LARGE, (curl_off_t) request.data->length()); + } + + if (request.verifyTLS) { + debug("verify TLS: Nix CA file = '%s'", settings.caFile); + if (settings.caFile != "") + curl_easy_setopt(req, CURLOPT_CAINFO, settings.caFile.c_str()); + } else { + curl_easy_setopt(req, CURLOPT_SSL_VERIFYPEER, 0); + curl_easy_setopt(req, CURLOPT_SSL_VERIFYHOST, 0); + } + + curl_easy_setopt(req, CURLOPT_CONNECTTIMEOUT, fileTransferSettings.connectTimeout.get()); + + curl_easy_setopt(req, CURLOPT_LOW_SPEED_LIMIT, 1L); + curl_easy_setopt(req, CURLOPT_LOW_SPEED_TIME, fileTransferSettings.stalledDownloadTimeout.get()); + + /* If no file exist in the specified path, curl continues to work + anyway as if netrc support was disabled. */ + curl_easy_setopt(req, CURLOPT_NETRC_FILE, settings.netrcFile.get().c_str()); + curl_easy_setopt(req, CURLOPT_NETRC, CURL_NETRC_OPTIONAL); + + if (writtenToSink) + curl_easy_setopt(req, CURLOPT_RESUME_FROM_LARGE, writtenToSink); + + result.data = std::make_shared(); + result.bodySize = 0; + } + + void finish(CURLcode code) + { + long httpStatus = 0; + curl_easy_getinfo(req, CURLINFO_RESPONSE_CODE, &httpStatus); + + char * effectiveUriCStr; + curl_easy_getinfo(req, CURLINFO_EFFECTIVE_URL, &effectiveUriCStr); + if (effectiveUriCStr) + result.effectiveUri = effectiveUriCStr; + + debug("finished %s of '%s'; curl status = %d, HTTP status = %d, body = %d bytes", + request.verb(), request.uri, code, httpStatus, result.bodySize); + + if (decompressionSink) { + try { + decompressionSink->finish(); + } catch (...) { + writeException = std::current_exception(); + } + } + + if (code == CURLE_WRITE_ERROR && result.etag == request.expectedETag) { + code = CURLE_OK; + httpStatus = 304; + } + + if (writeException) + failEx(writeException); + + else if (code == CURLE_OK && + (httpStatus == 200 || httpStatus == 201 || httpStatus == 204 || httpStatus == 206 || httpStatus == 304 || httpStatus == 226 /* FTP */ || httpStatus == 0 /* other protocol */)) + { + result.cached = httpStatus == 304; + act.progress(result.bodySize, result.bodySize); + done = true; + callback(std::move(result)); + } + + else { + // We treat most errors as transient, but won't retry when hopeless + Error err = Transient; + + if (httpStatus == 404 || httpStatus == 410 || code == CURLE_FILE_COULDNT_READ_FILE) { + // The file is definitely not there + err = NotFound; + } else if (httpStatus == 401 || httpStatus == 403 || httpStatus == 407) { + // Don't retry on authentication/authorization failures + err = Forbidden; + } else if (httpStatus >= 400 && httpStatus < 500 && httpStatus != 408 && httpStatus != 429) { + // Most 4xx errors are client errors and are probably not worth retrying: + // * 408 means the server timed out waiting for us, so we try again + // * 429 means too many requests, so we retry (with a delay) + err = Misc; + } else if (httpStatus == 501 || httpStatus == 505 || httpStatus == 511) { + // Let's treat most 5xx (server) errors as transient, except for a handful: + // * 501 not implemented + // * 505 http version not supported + // * 511 we're behind a captive portal + err = Misc; + } else { + // Don't bother retrying on certain cURL errors either + switch (code) { + case CURLE_FAILED_INIT: + case CURLE_URL_MALFORMAT: + case CURLE_NOT_BUILT_IN: + case CURLE_REMOTE_ACCESS_DENIED: + case CURLE_FILE_COULDNT_READ_FILE: + case CURLE_FUNCTION_NOT_FOUND: + case CURLE_ABORTED_BY_CALLBACK: + case CURLE_BAD_FUNCTION_ARGUMENT: + case CURLE_INTERFACE_FAILED: + case CURLE_UNKNOWN_OPTION: + case CURLE_SSL_CACERT_BADFILE: + case CURLE_TOO_MANY_REDIRECTS: + case CURLE_WRITE_ERROR: + case CURLE_UNSUPPORTED_PROTOCOL: + err = Misc; + break; + default: // Shut up warnings + break; + } + } + + attempt++; + + auto exc = + code == CURLE_ABORTED_BY_CALLBACK && _isInterrupted + ? FileTransferError(Interrupted, fmt("%s of '%s' was interrupted", request.verb(), request.uri)) + : httpStatus != 0 + ? FileTransferError(err, + fmt("unable to %s '%s': HTTP error %d", + request.verb(), request.uri, httpStatus) + + (code == CURLE_OK ? "" : fmt(" (curl error: %s)", curl_easy_strerror(code))) + ) + : FileTransferError(err, + fmt("unable to %s '%s': %s (%d)", + request.verb(), request.uri, curl_easy_strerror(code), code)); + + /* If this is a transient error, then maybe retry the + download after a while. If we're writing to a + sink, we can only retry if the server supports + ranged requests. */ + if (err == Transient + && attempt < request.tries + && (!this->request.dataCallback + || writtenToSink == 0 + || (acceptRanges && encoding.empty()))) + { + int ms = request.baseRetryTimeMs * std::pow(2.0f, attempt - 1 + std::uniform_real_distribution<>(0.0, 0.5)(fileTransfer.mt19937)); + if (writtenToSink) + warn("%s; retrying from offset %d in %d ms", exc.what(), writtenToSink, ms); + else + warn("%s; retrying in %d ms", exc.what(), ms); + embargo = std::chrono::steady_clock::now() + std::chrono::milliseconds(ms); + fileTransfer.enqueueItem(shared_from_this()); + } + else + fail(exc); + } + } + }; + + struct State + { + struct EmbargoComparator { + bool operator() (const std::shared_ptr & i1, const std::shared_ptr & i2) { + return i1->embargo > i2->embargo; + } + }; + bool quit = false; + std::priority_queue, std::vector>, EmbargoComparator> incoming; + }; + + Sync state_; + + /* We can't use a std::condition_variable to wake up the curl + thread, because it only monitors file descriptors. So use a + pipe instead. */ + Pipe wakeupPipe; + + std::thread workerThread; + + curlFileTransfer() + : mt19937(rd()) + { + static std::once_flag globalInit; + std::call_once(globalInit, curl_global_init, CURL_GLOBAL_ALL); + + curlm = curl_multi_init(); + + #if LIBCURL_VERSION_NUM >= 0x072b00 // Multiplex requires >= 7.43.0 + curl_multi_setopt(curlm, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX); + #endif + #if LIBCURL_VERSION_NUM >= 0x071e00 // Max connections requires >= 7.30.0 + curl_multi_setopt(curlm, CURLMOPT_MAX_TOTAL_CONNECTIONS, + fileTransferSettings.httpConnections.get()); + #endif + + wakeupPipe.create(); + fcntl(wakeupPipe.readSide.get(), F_SETFL, O_NONBLOCK); + + workerThread = std::thread([&]() { workerThreadEntry(); }); + } + + ~curlFileTransfer() + { + stopWorkerThread(); + + workerThread.join(); + + if (curlm) curl_multi_cleanup(curlm); + } + + void stopWorkerThread() + { + /* Signal the worker thread to exit. */ + { + auto state(state_.lock()); + state->quit = true; + } + writeFull(wakeupPipe.writeSide.get(), " ", false); + } + + void workerThreadMain() + { + /* Cause this thread to be notified on SIGINT. */ + auto callback = createInterruptCallback([&]() { + stopWorkerThread(); + }); + + std::map> items; + + bool quit = false; + + std::chrono::steady_clock::time_point nextWakeup; + + while (!quit) { + checkInterrupt(); + + /* Let curl do its thing. */ + int running; + CURLMcode mc = curl_multi_perform(curlm, &running); + if (mc != CURLM_OK) + throw nix::Error(format("unexpected error from curl_multi_perform(): %s") % curl_multi_strerror(mc)); + + /* Set the promises of any finished requests. */ + CURLMsg * msg; + int left; + while ((msg = curl_multi_info_read(curlm, &left))) { + if (msg->msg == CURLMSG_DONE) { + auto i = items.find(msg->easy_handle); + assert(i != items.end()); + i->second->finish(msg->data.result); + curl_multi_remove_handle(curlm, i->second->req); + i->second->active = false; + items.erase(i); + } + } + + /* Wait for activity, including wakeup events. */ + int numfds = 0; + struct curl_waitfd extraFDs[1]; + extraFDs[0].fd = wakeupPipe.readSide.get(); + extraFDs[0].events = CURL_WAIT_POLLIN; + extraFDs[0].revents = 0; + long maxSleepTimeMs = items.empty() ? 10000 : 100; + auto sleepTimeMs = + nextWakeup != std::chrono::steady_clock::time_point() + ? std::max(0, (int) std::chrono::duration_cast(nextWakeup - std::chrono::steady_clock::now()).count()) + : maxSleepTimeMs; + vomit("download thread waiting for %d ms", sleepTimeMs); + mc = curl_multi_wait(curlm, extraFDs, 1, sleepTimeMs, &numfds); + if (mc != CURLM_OK) + throw nix::Error(format("unexpected error from curl_multi_wait(): %s") % curl_multi_strerror(mc)); + + nextWakeup = std::chrono::steady_clock::time_point(); + + /* Add new curl requests from the incoming requests queue, + except for requests that are embargoed (waiting for a + retry timeout to expire). */ + if (extraFDs[0].revents & CURL_WAIT_POLLIN) { + char buf[1024]; + auto res = read(extraFDs[0].fd, buf, sizeof(buf)); + if (res == -1 && errno != EINTR) + throw SysError("reading curl wakeup socket"); + } + + std::vector> incoming; + auto now = std::chrono::steady_clock::now(); + + { + auto state(state_.lock()); + while (!state->incoming.empty()) { + auto item = state->incoming.top(); + if (item->embargo <= now) { + incoming.push_back(item); + state->incoming.pop(); + } else { + if (nextWakeup == std::chrono::steady_clock::time_point() + || item->embargo < nextWakeup) + nextWakeup = item->embargo; + break; + } + } + quit = state->quit; + } + + for (auto & item : incoming) { + debug("starting %s of %s", item->request.verb(), item->request.uri); + item->init(); + curl_multi_add_handle(curlm, item->req); + item->active = true; + items[item->req] = item; + } + } + + debug("download thread shutting down"); + } + + void workerThreadEntry() + { + try { + workerThreadMain(); + } catch (nix::Interrupted & e) { + } catch (std::exception & e) { + printError("unexpected error in download thread: %s", e.what()); + } + + { + auto state(state_.lock()); + while (!state->incoming.empty()) state->incoming.pop(); + state->quit = true; + } + } + + void enqueueItem(std::shared_ptr item) + { + if (item->request.data + && !hasPrefix(item->request.uri, "http://") + && !hasPrefix(item->request.uri, "https://")) + throw nix::Error("uploading to '%s' is not supported", item->request.uri); + + { + auto state(state_.lock()); + if (state->quit) + throw nix::Error("cannot enqueue download request because the download thread is shutting down"); + state->incoming.push(item); + } + writeFull(wakeupPipe.writeSide.get(), " "); + } + +#ifdef ENABLE_S3 + std::tuple parseS3Uri(std::string uri) + { + auto [path, params] = splitUriAndParams(uri); + + auto slash = path.find('/', 5); // 5 is the length of "s3://" prefix + if (slash == std::string::npos) + throw nix::Error("bad S3 URI '%s'", path); + + std::string bucketName(path, 5, slash - 5); + std::string key(path, slash + 1); + + return {bucketName, key, params}; + } +#endif + + void enqueueFileTransfer(const FileTransferRequest & request, + Callback callback) override + { + /* Ugly hack to support s3:// URIs. */ + if (hasPrefix(request.uri, "s3://")) { + // FIXME: do this on a worker thread + try { +#ifdef ENABLE_S3 + auto [bucketName, key, params] = parseS3Uri(request.uri); + + std::string profile = get(params, "profile").value_or(""); + std::string region = get(params, "region").value_or(Aws::Region::US_EAST_1); + std::string scheme = get(params, "scheme").value_or(""); + std::string endpoint = get(params, "endpoint").value_or(""); + + S3Helper s3Helper(profile, region, scheme, endpoint); + + // FIXME: implement ETag + auto s3Res = s3Helper.getObject(bucketName, key); + FileTransferResult res; + if (!s3Res.data) + throw FileTransferError(NotFound, fmt("S3 object '%s' does not exist", request.uri)); + res.data = s3Res.data; + callback(std::move(res)); +#else + throw nix::Error("cannot download '%s' because Nix is not built with S3 support", request.uri); +#endif + } catch (...) { callback.rethrow(); } + return; + } + + enqueueItem(std::make_shared(*this, request, std::move(callback))); + } +}; + +ref getFileTransfer() +{ + static ref fileTransfer = makeFileTransfer(); + return fileTransfer; +} + +ref makeFileTransfer() +{ + return make_ref(); +} + +std::future FileTransfer::enqueueFileTransfer(const FileTransferRequest & request) +{ + auto promise = std::make_shared>(); + enqueueFileTransfer(request, + {[promise](std::future fut) { + try { + promise->set_value(fut.get()); + } catch (...) { + promise->set_exception(std::current_exception()); + } + }}); + return promise->get_future(); +} + +FileTransferResult FileTransfer::download(const FileTransferRequest & request) +{ + return enqueueFileTransfer(request).get(); +} + +FileTransferResult FileTransfer::upload(const FileTransferRequest & request) +{ + /* Note: this method is the same as download, but helps in readability */ + return enqueueFileTransfer(request).get(); +} + +void FileTransfer::download(FileTransferRequest && request, Sink & sink) +{ + /* Note: we can't call 'sink' via request.dataCallback, because + that would cause the sink to execute on the fileTransfer + thread. If 'sink' is a coroutine, this will fail. Also, if the + sink is expensive (e.g. one that does decompression and writing + to the Nix store), it would stall the download thread too much. + Therefore we use a buffer to communicate data between the + download thread and the calling thread. */ + + struct State { + bool quit = false; + std::exception_ptr exc; + std::string data; + std::condition_variable avail, request; + }; + + auto _state = std::make_shared>(); + + /* In case of an exception, wake up the download thread. FIXME: + abort the download request. */ + Finally finally([&]() { + auto state(_state->lock()); + state->quit = true; + state->request.notify_one(); + }); + + request.dataCallback = [_state](char * buf, size_t len) { + + auto state(_state->lock()); + + if (state->quit) return; + + /* If the buffer is full, then go to sleep until the calling + thread wakes us up (i.e. when it has removed data from the + buffer). We don't wait forever to prevent stalling the + download thread. (Hopefully sleeping will throttle the + sender.) */ + if (state->data.size() > 1024 * 1024) { + debug("download buffer is full; going to sleep"); + state.wait_for(state->request, std::chrono::seconds(10)); + } + + /* Append data to the buffer and wake up the calling + thread. */ + state->data.append(buf, len); + state->avail.notify_one(); + }; + + enqueueFileTransfer(request, + {[_state](std::future fut) { + auto state(_state->lock()); + state->quit = true; + try { + fut.get(); + } catch (...) { + state->exc = std::current_exception(); + } + state->avail.notify_one(); + state->request.notify_one(); + }}); + + while (true) { + checkInterrupt(); + + std::string chunk; + + /* Grab data if available, otherwise wait for the download + thread to wake us up. */ + { + auto state(_state->lock()); + + while (state->data.empty()) { + + if (state->quit) { + if (state->exc) std::rethrow_exception(state->exc); + return; + } + + state.wait(state->avail); + } + + chunk = std::move(state->data); + + state->request.notify_one(); + } + + /* Flush the data to the sink and wake up the download thread + if it's blocked on a full buffer. We don't hold the state + lock while doing this to prevent blocking the download + thread if sink() takes a long time. */ + sink((unsigned char *) chunk.data(), chunk.size()); + } +} + +bool isUri(const string & s) +{ + if (s.compare(0, 8, "channel:") == 0) return true; + size_t pos = s.find("://"); + if (pos == string::npos) return false; + string scheme(s, 0, pos); + return scheme == "http" || scheme == "https" || scheme == "file" || scheme == "channel" || scheme == "git" || scheme == "s3" || scheme == "ssh"; +} + + +} diff --git a/src/libstore/filetransfer.hh b/src/libstore/filetransfer.hh new file mode 100644 index 000000000..2347f363d --- /dev/null +++ b/src/libstore/filetransfer.hh @@ -0,0 +1,116 @@ +#pragma once + +#include "types.hh" +#include "hash.hh" +#include "config.hh" + +#include +#include + +namespace nix { + +struct FileTransferSettings : Config +{ + Setting enableHttp2{this, true, "http2", + "Whether to enable HTTP/2 support."}; + + Setting userAgentSuffix{this, "", "user-agent-suffix", + "String appended to the user agent in HTTP requests."}; + + Setting httpConnections{this, 25, "http-connections", + "Number of parallel HTTP connections.", + {"binary-caches-parallel-connections"}}; + + Setting connectTimeout{this, 0, "connect-timeout", + "Timeout for connecting to servers during downloads. 0 means use curl's builtin default."}; + + Setting stalledDownloadTimeout{this, 300, "stalled-download-timeout", + "Timeout (in seconds) for receiving data from servers during download. Nix cancels idle downloads after this timeout's duration."}; + + Setting tries{this, 5, "download-attempts", + "How often Nix will attempt to download a file before giving up."}; +}; + +extern FileTransferSettings fileTransferSettings; + +struct FileTransferRequest +{ + std::string uri; + std::string expectedETag; + bool verifyTLS = true; + bool head = false; + size_t tries = fileTransferSettings.tries; + unsigned int baseRetryTimeMs = 250; + ActivityId parentAct; + bool decompress = true; + std::shared_ptr data; + std::string mimeType; + std::function dataCallback; + + FileTransferRequest(const std::string & uri) + : uri(uri), parentAct(getCurActivity()) { } + + std::string verb() + { + return data ? "upload" : "download"; + } +}; + +struct FileTransferResult +{ + bool cached = false; + std::string etag; + std::string effectiveUri; + std::shared_ptr data; + uint64_t bodySize = 0; +}; + +class Store; + +struct FileTransfer +{ + virtual ~FileTransfer() { } + + /* Enqueue a data transfer request, returning a future to the result of + the download. The future may throw a FileTransferError + exception. */ + virtual void enqueueFileTransfer(const FileTransferRequest & request, + Callback callback) = 0; + + std::future enqueueFileTransfer(const FileTransferRequest & request); + + /* Synchronously download a file. */ + FileTransferResult download(const FileTransferRequest & request); + + /* Synchronously upload a file. */ + FileTransferResult upload(const FileTransferRequest & request); + + /* Download a file, writing its data to a sink. The sink will be + invoked on the thread of the caller. */ + void download(FileTransferRequest && request, Sink & sink); + + enum Error { NotFound, Forbidden, Misc, Transient, Interrupted }; +}; + +/* Return a shared FileTransfer object. Using this object is preferred + because it enables connection reuse and HTTP/2 multiplexing. */ +ref getFileTransfer(); + +/* Return a new FileTransfer object. */ +ref makeFileTransfer(); + +class FileTransferError : public Error +{ +public: + FileTransfer::Error error; + FileTransferError(FileTransfer::Error error, const FormatOrString & fs) + : Error(fs), error(error) + { } +}; + +bool isUri(const string & s); + +/* Resolve deprecated 'channel:' URLs. */ +std::string resolveUri(const std::string & uri); + +} diff --git a/src/libstore/http-binary-cache-store.cc b/src/libstore/http-binary-cache-store.cc index ce0c5af3b..451a64785 100644 --- a/src/libstore/http-binary-cache-store.cc +++ b/src/libstore/http-binary-cache-store.cc @@ -1,5 +1,5 @@ #include "binary-cache-store.hh" -#include "datatransfer.hh" +#include "filetransfer.hh" #include "globals.hh" #include "nar-info-disk-cache.hh" diff --git a/src/libstore/s3-binary-cache-store.cc b/src/libstore/s3-binary-cache-store.cc index fccf010a4..b24e7b7d6 100644 --- a/src/libstore/s3-binary-cache-store.cc +++ b/src/libstore/s3-binary-cache-store.cc @@ -6,7 +6,7 @@ #include "nar-info-disk-cache.hh" #include "globals.hh" #include "compression.hh" -#include "datatransfer.hh" +#include "filetransfer.hh" #include "istringstream_nocopy.hh" #include diff --git a/src/nix-channel/nix-channel.cc b/src/nix-channel/nix-channel.cc index 0f8024d7b..7d584f891 100755 --- a/src/nix-channel/nix-channel.cc +++ b/src/nix-channel/nix-channel.cc @@ -1,6 +1,6 @@ #include "shared.hh" #include "globals.hh" -#include "datatransfer.hh" +#include "filetransfer.hh" #include "store-api.hh" #include "../nix/legacy.hh" #include "fetchers.hh" diff --git a/src/nix-prefetch-url/nix-prefetch-url.cc b/src/nix-prefetch-url/nix-prefetch-url.cc index e136bb95a..748554b9c 100644 --- a/src/nix-prefetch-url/nix-prefetch-url.cc +++ b/src/nix-prefetch-url/nix-prefetch-url.cc @@ -1,6 +1,6 @@ #include "hash.hh" #include "shared.hh" -#include "datatransfer.hh" +#include "filetransfer.hh" #include "store-api.hh" #include "eval.hh" #include "eval-inline.hh" diff --git a/src/nix/main.cc b/src/nix/main.cc index 140c963e0..2c64c7476 100644 --- a/src/nix/main.cc +++ b/src/nix/main.cc @@ -8,7 +8,7 @@ #include "shared.hh" #include "store-api.hh" #include "progress-bar.hh" -#include "datatransfer.hh" +#include "filetransfer.hh" #include "finally.hh" #include diff --git a/src/nix/upgrade-nix.cc b/src/nix/upgrade-nix.cc index 574f7d8c2..4fcc6a721 100644 --- a/src/nix/upgrade-nix.cc +++ b/src/nix/upgrade-nix.cc @@ -1,7 +1,7 @@ #include "command.hh" #include "common-args.hh" #include "store-api.hh" -#include "datatransfer.hh" +#include "filetransfer.hh" #include "eval.hh" #include "attr-path.hh" #include "names.hh" -- cgit v1.2.3 From 7867685dcdf1ba3c9290b777e0975ff0b775f667 Mon Sep 17 00:00:00 2001 From: Nikola Knezevic Date: Wed, 8 Apr 2020 14:12:22 +0200 Subject: after flake rebase --- src/libexpr/parser.y | 2 +- src/libexpr/primops/fetchTree.cc | 2 +- src/libfetchers/github.cc | 2 +- src/libfetchers/tarball.cc | 10 +++++----- src/nix-channel/nix-channel.cc | 2 +- 5 files changed, 9 insertions(+), 9 deletions(-) (limited to 'src') diff --git a/src/libexpr/parser.y b/src/libexpr/parser.y index 235ed5b1d..b33ab908b 100644 --- a/src/libexpr/parser.y +++ b/src/libexpr/parser.y @@ -690,7 +690,7 @@ std::pair EvalState::resolveSearchPathElem(const SearchPathEl try { res = { true, store->toRealPath(fetchers::downloadTarball( store, resolveUri(elem.second), "source", false).storePath) }; - } catch (DataTransferError & e) { + } catch (FileTransferError & e) { printError(format("warning: Nix search path entry '%1%' cannot be downloaded, ignoring") % elem.second); res = { false, "" }; } diff --git a/src/libexpr/primops/fetchTree.cc b/src/libexpr/primops/fetchTree.cc index 9586f71ed..43c58485a 100644 --- a/src/libexpr/primops/fetchTree.cc +++ b/src/libexpr/primops/fetchTree.cc @@ -2,7 +2,7 @@ #include "eval-inline.hh" #include "store-api.hh" #include "fetchers.hh" -#include "download.hh" +#include "filetransfer.hh" #include #include diff --git a/src/libfetchers/github.cc b/src/libfetchers/github.cc index ef27eaa76..8675a5a66 100644 --- a/src/libfetchers/github.cc +++ b/src/libfetchers/github.cc @@ -1,4 +1,4 @@ -#include "download.hh" +#include "filetransfer.hh" #include "cache.hh" #include "fetchers.hh" #include "globals.hh" diff --git a/src/libfetchers/tarball.cc b/src/libfetchers/tarball.cc index 4c4e5828e..695525b31 100644 --- a/src/libfetchers/tarball.cc +++ b/src/libfetchers/tarball.cc @@ -1,6 +1,6 @@ #include "fetchers.hh" #include "cache.hh" -#include "download.hh" +#include "filetransfer.hh" #include "globals.hh" #include "store-api.hh" #include "archive.hh" @@ -36,13 +36,13 @@ DownloadFileResult downloadFile( if (cached && !cached->expired) return useCached(); - DownloadRequest request(url); + FileTransferRequest request(url); if (cached) request.expectedETag = getStrAttr(cached->infoAttrs, "etag"); - DownloadResult res; + FileTransferResult res; try { - res = getDownloader()->download(request); - } catch (DownloadError & e) { + res = getFileTransfer()->download(request); + } catch (FileTransferError & e) { if (cached) { warn("%s; using cached version", e.msg()); return useCached(); diff --git a/src/nix-channel/nix-channel.cc b/src/nix-channel/nix-channel.cc index 7d584f891..abd390414 100755 --- a/src/nix-channel/nix-channel.cc +++ b/src/nix-channel/nix-channel.cc @@ -113,7 +113,7 @@ static void update(const StringSet & channelNames) // Download the channel tarball. try { filename = store->toRealPath(fetchers::downloadFile(store, url + "/nixexprs.tar.xz", "nixexprs.tar.xz", false).storePath); - } catch (DataTransferError & e) { + } catch (FileTransferError & e) { filename = store->toRealPath(fetchers::downloadFile(store, url + "/nixexprs.tar.bz2", "nixexprs.tar.bz2", false).storePath); } } -- cgit v1.2.3 From a693a9fa4b92df0d3283a9de8b1e968378a026d2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Domen=20Ko=C5=BEar?= Date: Thu, 9 Apr 2020 09:45:15 +0200 Subject: Attach pos to if expression errors --- src/libexpr/eval.cc | 2 +- src/libexpr/nixexpr.hh | 3 ++- src/libexpr/parser.y | 2 +- 3 files changed, 4 insertions(+), 3 deletions(-) (limited to 'src') diff --git a/src/libexpr/eval.cc b/src/libexpr/eval.cc index b91a021b4..f963a42ca 100644 --- a/src/libexpr/eval.cc +++ b/src/libexpr/eval.cc @@ -1256,7 +1256,7 @@ void ExprWith::eval(EvalState & state, Env & env, Value & v) void ExprIf::eval(EvalState & state, Env & env, Value & v) { - (state.evalBool(env, cond) ? then : else_)->eval(state, env, v); + (state.evalBool(env, cond, pos) ? then : else_)->eval(state, env, v); } diff --git a/src/libexpr/nixexpr.hh b/src/libexpr/nixexpr.hh index 8c96de37c..25798cac6 100644 --- a/src/libexpr/nixexpr.hh +++ b/src/libexpr/nixexpr.hh @@ -262,8 +262,9 @@ struct ExprWith : Expr struct ExprIf : Expr { + Pos pos; Expr * cond, * then, * else_; - ExprIf(Expr * cond, Expr * then, Expr * else_) : cond(cond), then(then), else_(else_) { }; + ExprIf(const Pos & pos, Expr * cond, Expr * then, Expr * else_) : pos(pos), cond(cond), then(then), else_(else_) { }; COMMON_METHODS }; diff --git a/src/libexpr/parser.y b/src/libexpr/parser.y index b33ab908b..1993fa6c1 100644 --- a/src/libexpr/parser.y +++ b/src/libexpr/parser.y @@ -335,7 +335,7 @@ expr_function ; expr_if - : IF expr THEN expr ELSE expr { $$ = new ExprIf($2, $4, $6); } + : IF expr THEN expr ELSE expr { $$ = new ExprIf(CUR_POS, $2, $4, $6); } | expr_op ; -- cgit v1.2.3