lix/src/libstore/download.cc

668 lines
24 KiB
C++
Raw Normal View History

2015-04-09 10:12:50 +00:00
#include "download.hh"
#include "util.hh"
#include "globals.hh"
#include "hash.hh"
#include "store-api.hh"
#include "archive.hh"
#include "s3.hh"
2015-04-09 10:12:50 +00:00
#include <unistd.h>
#include <fcntl.h>
2015-04-09 10:12:50 +00:00
#include <curl/curl.h>
2016-10-17 21:11:04 +00:00
#include <queue>
2015-10-07 15:31:50 +00:00
#include <iostream>
#include <thread>
#include <cmath>
#include <random>
2015-10-07 15:31:50 +00:00
2015-04-09 10:12:50 +00:00
namespace nix {
2015-10-07 15:31:50 +00:00
double getTime()
{
struct timeval tv;
gettimeofday(&tv, 0);
return tv.tv_sec + (tv.tv_usec / 1000000.0);
}
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
2015-04-09 10:12:50 +00:00
{
CURLM * curlm = 0;
2015-04-09 10:12:50 +00:00
std::random_device rd;
std::mt19937 mt19937;
2015-10-07 15:31:50 +00:00
bool enableHttp2;
struct DownloadItem : public std::enable_shared_from_this<DownloadItem>
2015-04-09 10:12:50 +00:00
{
CurlDownloader & downloader;
DownloadRequest request;
DownloadResult result;
bool done = false; // whether either the success or failure function has been called
std::function<void(const DownloadResult &)> success;
std::function<void(std::exception_ptr exc)> failure;
CURL * req = 0;
bool active = false; // whether the handle has been added to the multi object
std::string status;
bool showProgress = false;
double prevProgressTime{0}, startTime{0};
unsigned int moveBack{1};
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;
DownloadItem(CurlDownloader & downloader, const DownloadRequest & request)
: downloader(downloader), request(request)
{
showProgress =
request.showProgress == DownloadRequest::yes ||
(request.showProgress == DownloadRequest::automatic && isatty(STDERR_FILENO));
if (!request.expectedETag.empty())
requestHeaders = curl_slist_append(requestHeaders, ("If-None-Match: " + request.expectedETag).c_str());
}
2015-04-09 10:12:50 +00:00
~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();
}
}
template<class T>
void fail(const T & e)
{
assert(!done);
done = true;
2016-09-20 15:49:31 +00:00
callFailure(failure, std::make_exception_ptr(e));
}
size_t writeCallback(void * contents, size_t size, size_t nmemb)
{
size_t realSize = size * nmemb;
result.data->append((char *) contents, realSize);
return realSize;
}
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<vector<string>>(line, " ");
status = ss.size() >= 2 ? ss[1] : "";
result.data = std::make_shared<std::string>();
} 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;
}
2015-04-09 10:12:50 +00:00
}
}
}
return realSize;
2015-04-09 10:12:50 +00:00
}
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)
{
if (showProgress) {
double now = getTime();
if (prevProgressTime <= now - 1) {
string s = (format(" [%1$.0f/%2$.0f KiB, %3$.1f KiB/s]")
% (dlnow / 1024.0)
% (dltotal / 1024.0)
% (now == startTime ? 0 : dlnow / 1024.0 / (now - startTime))).str();
std::cerr << "\e[" << moveBack << "D" << s;
moveBack = s.size();
std::cerr.flush();
prevProgressTime = now;
}
}
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;
}
void init()
{
// FIXME: handle parallel downloads.
if (showProgress) {
std::cerr << (format("downloading %1%... ") % request.uri);
2015-10-07 15:31:50 +00:00
std::cerr.flush();
startTime = getTime();
2015-10-07 15:31:50 +00:00
}
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_NOSIGNAL, 1);
curl_easy_setopt(req, CURLOPT_USERAGENT, ("curl/" LIBCURL_VERSION " Nix/" + nixVersion).c_str());
#if LIBCURL_VERSION_NUM >= 0x072b00
curl_easy_setopt(req, CURLOPT_PIPEWAIT, 1);
#endif
#if LIBCURL_VERSION_NUM >= 0x072f00
if (downloader.enableHttp2)
curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS);
#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.verifyTLS)
curl_easy_setopt(req, CURLOPT_CAINFO,
getEnv("NIX_SSL_CERT_FILE", getEnv("SSL_CERT_FILE", "/etc/ssl/certs/ca-certificates.crt")).c_str());
else {
curl_easy_setopt(req, CURLOPT_SSL_VERIFYPEER, 0);
curl_easy_setopt(req, CURLOPT_SSL_VERIFYHOST, 0);
}
2017-02-09 17:16:09 +00:00
/* 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.c_str());
2017-02-01 12:37:34 +00:00
curl_easy_setopt(req, CURLOPT_NETRC, CURL_NETRC_OPTIONAL);
result.data = std::make_shared<std::string>();
2015-10-07 15:31:50 +00:00
}
2015-05-05 12:39:48 +00:00
void finish(CURLcode code)
{
if (showProgress)
//std::cerr << "\e[" << moveBack << "D\e[K\n";
std::cerr << "\n";
2015-10-07 15:31:50 +00:00
long httpStatus = 0;
curl_easy_getinfo(req, CURLINFO_RESPONSE_CODE, &httpStatus);
2015-04-09 10:12:50 +00:00
char * effectiveUrlCStr;
curl_easy_getinfo(req, CURLINFO_EFFECTIVE_URL, &effectiveUrlCStr);
if (effectiveUrlCStr)
result.effectiveUrl = effectiveUrlCStr;
debug(format("finished download of %s; curl status = %d, HTTP status = %d, body = %d bytes")
% request.uri % code % httpStatus % (result.data ? result.data->size() : 0));
if (code == CURLE_WRITE_ERROR && result.etag == request.expectedETag) {
code = CURLE_OK;
httpStatus = 304;
}
if (code == CURLE_OK &&
(httpStatus == 200 || httpStatus == 304 || httpStatus == 226 /* FTP */ || httpStatus == 0 /* other protocol */))
{
result.cached = httpStatus == 304;
done = true;
2016-09-20 15:25:44 +00:00
callSuccess(success, failure, const_cast<const DownloadResult &>(result));
} else {
Error err =
(httpStatus == 404 || code == CURLE_FILE_COULDNT_READ_FILE) ? NotFound :
httpStatus == 403 ? Forbidden :
(httpStatus == 408 || httpStatus == 500 || httpStatus == 503
|| httpStatus == 504 || httpStatus == 522 || httpStatus == 524
|| code == CURLE_COULDNT_RESOLVE_HOST || code == CURLE_RECV_ERROR) ? Transient :
Misc;
attempt++;
2015-04-09 10:12:50 +00:00
auto exc =
code == CURLE_ABORTED_BY_CALLBACK && _isInterrupted
? DownloadError(Interrupted, format("download of %s was interrupted") % request.uri)
: httpStatus != 0
? DownloadError(err, format("unable to download %s: HTTP error %d (curl error: %s)") % request.uri % httpStatus % curl_easy_strerror(code))
: DownloadError(err, format("unable to download %s: %s (%d)") % request.uri % curl_easy_strerror(code) % code);
/* If this is a transient error, then maybe retry the
download after a while. */
if (err == Transient && attempt < request.tries) {
int ms = request.baseRetryTimeMs * std::pow(2.0f, attempt - 1 + std::uniform_real_distribution<>(0.0, 0.5)(downloader.mt19937));
printError(format("warning: %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);
}
}
};
2015-04-09 10:12:50 +00:00
struct State
{
2016-10-17 21:11:04 +00:00
struct EmbargoComparator {
bool operator() (const std::shared_ptr<DownloadItem> & i1, const std::shared_ptr<DownloadItem> & i2) {
return i1->embargo > i2->embargo;
}
};
bool quit = false;
2016-10-17 21:11:04 +00:00
std::priority_queue<std::shared_ptr<DownloadItem>, std::vector<std::shared_ptr<DownloadItem>>, EmbargoComparator> incoming;
};
2015-04-09 10:12:50 +00:00
Sync<State> state_;
2015-05-05 12:39:48 +00:00
/* 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()
2016-10-05 19:29:18 +00:00
: mt19937(rd())
{
static std::once_flag globalInit;
std::call_once(globalInit, curl_global_init, CURL_GLOBAL_ALL);
2015-04-09 10:12:50 +00:00
curlm = curl_multi_init();
#if LIBCURL_VERSION_NUM >= 0x072b00 // correct?
curl_multi_setopt(curlm, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX);
#endif
curl_multi_setopt(curlm, CURLMOPT_MAX_TOTAL_CONNECTIONS,
settings.binaryCachesParallelConnections);
enableHttp2 = settings.enableHttp2;
2015-04-09 10:12:50 +00:00
wakeupPipe.create();
fcntl(wakeupPipe.readSide.get(), F_SETFL, O_NONBLOCK);
workerThread = std::thread([&]() { workerThreadEntry(); });
}
2015-04-09 10:12:50 +00:00
~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;
2015-04-09 10:12:50 +00:00
}
writeFull(wakeupPipe.writeSide.get(), " ", false);
}
void workerThreadMain()
{
/* Cause this thread to be notified on SIGINT. */
auto callback = createInterruptCallback([&]() {
stopWorkerThread();
});
std::map<CURL *, std::shared_ptr<DownloadItem>> items;
2016-10-19 13:02:38 +00:00
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);
}
}
2015-10-07 15:31:50 +00:00
/* 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;
auto sleepTimeMs =
nextWakeup != std::chrono::steady_clock::time_point()
? std::max(0, (int) std::chrono::duration_cast<std::chrono::milliseconds>(nextWakeup - std::chrono::steady_clock::now()).count())
: 1000000000;
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
2016-10-17 21:11:04 +00:00
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");
}
2015-04-09 10:12:50 +00:00
2016-10-17 21:11:04 +00:00
std::vector<std::shared_ptr<DownloadItem>> incoming;
auto now = std::chrono::steady_clock::now();
{
auto state(state_.lock());
2016-10-17 21:11:04 +00:00
while (!state->incoming.empty()) {
auto item = state->incoming.top();
if (item->embargo <= now) {
incoming.push_back(item);
2016-10-17 21:11:04 +00:00
state->incoming.pop();
} else {
if (nextWakeup == std::chrono::steady_clock::time_point()
|| item->embargo < nextWakeup)
nextWakeup = item->embargo;
2016-10-17 21:11:04 +00:00
break;
}
}
quit = state->quit;
}
2016-08-11 15:34:43 +00:00
for (auto & item : incoming) {
debug(format("starting download of %s") % item->request.uri);
item->init();
curl_multi_add_handle(curlm, item->req);
item->active = true;
items[item->req] = item;
}
}
2015-04-09 10:12:50 +00:00
debug("download thread shutting down");
2015-04-09 10:12:50 +00:00
}
void workerThreadEntry()
{
try {
workerThreadMain();
} catch (nix::Interrupted & e) {
} catch (std::exception & e) {
printError(format("unexpected error in download thread: %s") % e.what());
}
{
auto state(state_.lock());
2016-10-17 21:11:04 +00:00
while (!state->incoming.empty()) state->incoming.pop();
state->quit = true;
}
}
void enqueueItem(std::shared_ptr<DownloadItem> item)
{
{
auto state(state_.lock());
if (state->quit)
throw nix::Error("cannot enqueue download request because the download thread is shutting down");
2016-10-17 21:11:04 +00:00
state->incoming.push(item);
}
writeFull(wakeupPipe.writeSide.get(), " ");
}
void enqueueDownload(const DownloadRequest & request,
std::function<void(const DownloadResult &)> success,
std::function<void(std::exception_ptr exc)> failure) override
{
/* Ugly hack to support s3:// URIs. */
if (hasPrefix(request.uri, "s3://")) {
// FIXME: do this on a worker thread
sync2async<DownloadResult>(success, failure, [&]() -> DownloadResult {
#ifdef ENABLE_S3
S3Helper s3Helper;
auto slash = request.uri.find('/', 5);
if (slash == std::string::npos)
throw nix::Error("bad S3 URI %s", request.uri);
std::string bucketName(request.uri, 5, slash - 5);
std::string key(request.uri, slash + 1);
// 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;
return res;
#else
throw nix::Error("cannot download %s because Nix is not built with S3 support", request.uri);
#endif
});
return;
}
auto item = std::make_shared<DownloadItem>(*this, request);
item->success = success;
item->failure = failure;
enqueueItem(item);
}
2015-04-09 10:12:50 +00:00
};
ref<Downloader> getDownloader()
{
static std::shared_ptr<Downloader> downloader;
static std::once_flag downloaderCreated;
std::call_once(downloaderCreated, [&]() { downloader = makeDownloader(); });
return ref<Downloader>(downloader);
}
ref<Downloader> makeDownloader()
2015-04-09 10:12:50 +00:00
{
return make_ref<CurlDownloader>();
2015-04-09 10:12:50 +00:00
}
std::future<DownloadResult> Downloader::enqueueDownload(const DownloadRequest & request)
{
auto promise = std::make_shared<std::promise<DownloadResult>>();
enqueueDownload(request,
[promise](const DownloadResult & result) { promise->set_value(result); },
[promise](std::exception_ptr exc) { promise->set_exception(exc); });
return promise->get_future();
}
DownloadResult Downloader::download(const DownloadRequest & request)
{
return enqueueDownload(request).get();
}
Path Downloader::downloadCached(ref<Store> store, const string & url_, bool unpack, string name, const Hash & expectedHash, string * effectiveUrl)
{
auto url = resolveUri(url_);
if (name == "") {
auto p = url.rfind('/');
if (p != string::npos) name = string(url, p + 1);
}
Path expectedStorePath;
if (expectedHash) {
expectedStorePath = store->makeFixedOutputPath(unpack, expectedHash, name);
if (store->isValidPath(expectedStorePath))
return expectedStorePath;
}
Path cacheDir = getCacheDir() + "/nix/tarballs";
createDirs(cacheDir);
string urlHash = printHash32(hashString(htSHA256, url));
Path dataFile = cacheDir + "/" + urlHash + ".info";
Path fileLink = cacheDir + "/" + urlHash + "-file";
Path storePath;
string expectedETag;
int ttl = settings.tarballTtl;
bool skip = false;
if (pathExists(fileLink) && pathExists(dataFile)) {
storePath = readLink(fileLink);
store->addTempRoot(storePath);
if (store->isValidPath(storePath)) {
auto ss = tokenizeString<vector<string>>(readFile(dataFile), "\n");
if (ss.size() >= 3 && ss[0] == url) {
time_t lastChecked;
2016-08-11 15:34:43 +00:00
if (string2Int(ss[2], lastChecked) && lastChecked + ttl >= time(0)) {
skip = true;
if (effectiveUrl)
*effectiveUrl = url_;
2016-08-11 15:34:43 +00:00
} else if (!ss[1].empty()) {
debug(format("verifying previous ETag %1%") % ss[1]);
expectedETag = ss[1];
}
}
} else
storePath = "";
}
if (!skip) {
try {
DownloadRequest request(url);
request.expectedETag = expectedETag;
auto res = download(request);
if (effectiveUrl)
*effectiveUrl = res.effectiveUrl;
if (!res.cached) {
ValidPathInfo info;
StringSink sink;
dumpString(*res.data, sink);
Hash hash = hashString(expectedHash ? expectedHash.type : htSHA256, *res.data);
info.path = store->makeFixedOutputPath(false, hash, name);
info.narHash = hashString(htSHA256, *sink.s);
store->addToStore(info, sink.s, false, true);
storePath = info.path;
}
assert(!storePath.empty());
replaceSymlink(storePath, fileLink);
2015-10-29 12:26:55 +00:00
writeFile(dataFile, url + "\n" + res.etag + "\n" + std::to_string(time(0)) + "\n");
} catch (DownloadError & e) {
if (storePath.empty()) throw;
printError(format("warning: %1%; using cached result") % e.msg());
}
}
if (unpack) {
Path unpackedLink = cacheDir + "/" + baseNameOf(storePath) + "-unpacked";
Path unpackedStorePath;
if (pathExists(unpackedLink)) {
unpackedStorePath = readLink(unpackedLink);
store->addTempRoot(unpackedStorePath);
if (!store->isValidPath(unpackedStorePath))
unpackedStorePath = "";
}
if (unpackedStorePath.empty()) {
printInfo(format("unpacking %1%...") % url);
Path tmpDir = createTempDir();
AutoDelete autoDelete(tmpDir, true);
2015-06-01 13:14:44 +00:00
// FIXME: this requires GNU tar for decompression.
runProgram("tar", true, {"xf", storePath, "-C", tmpDir, "--strip-components", "1"}, "");
unpackedStorePath = store->addToStore(name, tmpDir, true, htSHA256, defaultPathFilter, false);
}
replaceSymlink(unpackedStorePath, unpackedLink);
storePath = unpackedStorePath;
}
if (expectedStorePath != "" && storePath != expectedStorePath)
throw nix::Error(format("hash mismatch in file downloaded from %s") % url);
return storePath;
}
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";
}
2015-04-09 10:12:50 +00:00
}