DataTransfer -> FileTransfer
This commit is contained in:
parent
7848372b0f
commit
c330109bfa
13 changed files with 107 additions and 107 deletions
|
@ -361,7 +361,7 @@ public:
|
||||||
{
|
{
|
||||||
actDerivations.progress(doneBuilds, expectedBuilds + doneBuilds, runningBuilds, failedBuilds);
|
actDerivations.progress(doneBuilds, expectedBuilds + doneBuilds, runningBuilds, failedBuilds);
|
||||||
actSubstitutions.progress(doneSubstitutions, expectedSubstitutions + doneSubstitutions, runningSubstitutions, failedSubstitutions);
|
actSubstitutions.progress(doneSubstitutions, expectedSubstitutions + doneSubstitutions, runningSubstitutions, failedSubstitutions);
|
||||||
act.setExpected(actDataTransfer, expectedDownloadSize + doneDownloadSize);
|
act.setExpected(actFileTransfer, expectedDownloadSize + doneDownloadSize);
|
||||||
act.setExpected(actCopyPath, expectedNarSize + doneNarSize);
|
act.setExpected(actCopyPath, expectedNarSize + doneNarSize);
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
|
@ -26,9 +26,9 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData)
|
||||||
auto mainUrl = getAttr("url");
|
auto mainUrl = getAttr("url");
|
||||||
bool unpack = get(drv.env, "unpack").value_or("") == "1";
|
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. */
|
a forked process. */
|
||||||
auto dataTransfer = makeDataTransfer();
|
auto fileTransfer = makeFileTransfer();
|
||||||
|
|
||||||
auto fetch = [&](const std::string & url) {
|
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
|
/* No need to do TLS verification, because we check the hash of
|
||||||
the result anyway. */
|
the result anyway. */
|
||||||
DataTransferRequest request(url);
|
FileTransferRequest request(url);
|
||||||
request.verifyTLS = false;
|
request.verifyTLS = false;
|
||||||
request.decompress = false;
|
request.decompress = false;
|
||||||
|
|
||||||
auto decompressor = makeDecompressionSink(
|
auto decompressor = makeDecompressionSink(
|
||||||
unpack && hasSuffix(mainUrl, ".xz") ? "xz" : "none", sink);
|
unpack && hasSuffix(mainUrl, ".xz") ? "xz" : "none", sink);
|
||||||
dataTransfer->download(std::move(request), *decompressor);
|
fileTransfer->download(std::move(request), *decompressor);
|
||||||
decompressor->finish();
|
decompressor->finish();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|
|
@ -27,9 +27,9 @@ using namespace std::string_literals;
|
||||||
|
|
||||||
namespace nix {
|
namespace nix {
|
||||||
|
|
||||||
DataTransferSettings dataTransferSettings;
|
FileTransferSettings fileTransferSettings;
|
||||||
|
|
||||||
static GlobalConfig::Register r1(&dataTransferSettings);
|
static GlobalConfig::Register r1(&fileTransferSettings);
|
||||||
|
|
||||||
std::string resolveUri(const std::string & uri)
|
std::string resolveUri(const std::string & uri)
|
||||||
{
|
{
|
||||||
|
@ -39,7 +39,7 @@ std::string resolveUri(const std::string & uri)
|
||||||
return uri;
|
return uri;
|
||||||
}
|
}
|
||||||
|
|
||||||
struct curlDataTransfer : public DataTransfer
|
struct curlFileTransfer : public FileTransfer
|
||||||
{
|
{
|
||||||
CURLM * curlm = 0;
|
CURLM * curlm = 0;
|
||||||
|
|
||||||
|
@ -48,12 +48,12 @@ struct curlDataTransfer : public DataTransfer
|
||||||
|
|
||||||
struct TransferItem : public std::enable_shared_from_this<TransferItem>
|
struct TransferItem : public std::enable_shared_from_this<TransferItem>
|
||||||
{
|
{
|
||||||
curlDataTransfer & dataTransfer;
|
curlFileTransfer & fileTransfer;
|
||||||
DataTransferRequest request;
|
FileTransferRequest request;
|
||||||
DataTransferResult result;
|
FileTransferResult result;
|
||||||
Activity act;
|
Activity act;
|
||||||
bool done = false; // whether either the success or failure function has been called
|
bool done = false; // whether either the success or failure function has been called
|
||||||
Callback<DataTransferResult> callback;
|
Callback<FileTransferResult> callback;
|
||||||
CURL * req = 0;
|
CURL * req = 0;
|
||||||
bool active = false; // whether the handle has been added to the multi object
|
bool active = false; // whether the handle has been added to the multi object
|
||||||
std::string status;
|
std::string status;
|
||||||
|
@ -72,12 +72,12 @@ struct curlDataTransfer : public DataTransfer
|
||||||
|
|
||||||
curl_off_t writtenToSink = 0;
|
curl_off_t writtenToSink = 0;
|
||||||
|
|
||||||
TransferItem(curlDataTransfer & dataTransfer,
|
TransferItem(curlFileTransfer & fileTransfer,
|
||||||
const DataTransferRequest & request,
|
const FileTransferRequest & request,
|
||||||
Callback<DataTransferResult> && callback)
|
Callback<FileTransferResult> && callback)
|
||||||
: dataTransfer(dataTransfer)
|
: fileTransfer(fileTransfer)
|
||||||
, request(request)
|
, request(request)
|
||||||
, act(*logger, lvlTalkative, actDataTransfer,
|
, act(*logger, lvlTalkative, actFileTransfer,
|
||||||
fmt(request.data ? "uploading '%s'" : "downloading '%s'", request.uri),
|
fmt(request.data ? "uploading '%s'" : "downloading '%s'", request.uri),
|
||||||
{request.uri}, request.parentAct)
|
{request.uri}, request.parentAct)
|
||||||
, callback(std::move(callback))
|
, callback(std::move(callback))
|
||||||
|
@ -106,13 +106,13 @@ struct curlDataTransfer : public DataTransfer
|
||||||
{
|
{
|
||||||
if (req) {
|
if (req) {
|
||||||
if (active)
|
if (active)
|
||||||
curl_multi_remove_handle(dataTransfer.curlm, req);
|
curl_multi_remove_handle(fileTransfer.curlm, req);
|
||||||
curl_easy_cleanup(req);
|
curl_easy_cleanup(req);
|
||||||
}
|
}
|
||||||
if (requestHeaders) curl_slist_free_all(requestHeaders);
|
if (requestHeaders) curl_slist_free_all(requestHeaders);
|
||||||
try {
|
try {
|
||||||
if (!done)
|
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 (...) {
|
} catch (...) {
|
||||||
ignoreException();
|
ignoreException();
|
||||||
}
|
}
|
||||||
|
@ -257,12 +257,12 @@ struct curlDataTransfer : public DataTransfer
|
||||||
curl_easy_setopt(req, CURLOPT_NOSIGNAL, 1);
|
curl_easy_setopt(req, CURLOPT_NOSIGNAL, 1);
|
||||||
curl_easy_setopt(req, CURLOPT_USERAGENT,
|
curl_easy_setopt(req, CURLOPT_USERAGENT,
|
||||||
("curl/" LIBCURL_VERSION " Nix/" + nixVersion +
|
("curl/" LIBCURL_VERSION " Nix/" + nixVersion +
|
||||||
(dataTransferSettings.userAgentSuffix != "" ? " " + dataTransferSettings.userAgentSuffix.get() : "")).c_str());
|
(fileTransferSettings.userAgentSuffix != "" ? " " + fileTransferSettings.userAgentSuffix.get() : "")).c_str());
|
||||||
#if LIBCURL_VERSION_NUM >= 0x072b00
|
#if LIBCURL_VERSION_NUM >= 0x072b00
|
||||||
curl_easy_setopt(req, CURLOPT_PIPEWAIT, 1);
|
curl_easy_setopt(req, CURLOPT_PIPEWAIT, 1);
|
||||||
#endif
|
#endif
|
||||||
#if LIBCURL_VERSION_NUM >= 0x072f00
|
#if LIBCURL_VERSION_NUM >= 0x072f00
|
||||||
if (dataTransferSettings.enableHttp2)
|
if (fileTransferSettings.enableHttp2)
|
||||||
curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS);
|
curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS);
|
||||||
else
|
else
|
||||||
curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1);
|
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_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_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
|
/* If no file exist in the specified path, curl continues to work
|
||||||
anyway as if netrc support was disabled. */
|
anyway as if netrc support was disabled. */
|
||||||
|
@ -401,14 +401,14 @@ struct curlDataTransfer : public DataTransfer
|
||||||
|
|
||||||
auto exc =
|
auto exc =
|
||||||
code == CURLE_ABORTED_BY_CALLBACK && _isInterrupted
|
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
|
: httpStatus != 0
|
||||||
? DataTransferError(err,
|
? FileTransferError(err,
|
||||||
fmt("unable to %s '%s': HTTP error %d",
|
fmt("unable to %s '%s': HTTP error %d",
|
||||||
request.verb(), request.uri, httpStatus)
|
request.verb(), request.uri, httpStatus)
|
||||||
+ (code == CURLE_OK ? "" : fmt(" (curl error: %s)", curl_easy_strerror(code)))
|
+ (code == CURLE_OK ? "" : fmt(" (curl error: %s)", curl_easy_strerror(code)))
|
||||||
)
|
)
|
||||||
: DataTransferError(err,
|
: FileTransferError(err,
|
||||||
fmt("unable to %s '%s': %s (%d)",
|
fmt("unable to %s '%s': %s (%d)",
|
||||||
request.verb(), request.uri, curl_easy_strerror(code), code));
|
request.verb(), request.uri, curl_easy_strerror(code), code));
|
||||||
|
|
||||||
|
@ -422,13 +422,13 @@ struct curlDataTransfer : public DataTransfer
|
||||||
|| writtenToSink == 0
|
|| writtenToSink == 0
|
||||||
|| (acceptRanges && encoding.empty())))
|
|| (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)
|
if (writtenToSink)
|
||||||
warn("%s; retrying from offset %d in %d ms", exc.what(), writtenToSink, ms);
|
warn("%s; retrying from offset %d in %d ms", exc.what(), writtenToSink, ms);
|
||||||
else
|
else
|
||||||
warn("%s; retrying in %d ms", exc.what(), ms);
|
warn("%s; retrying in %d ms", exc.what(), ms);
|
||||||
embargo = std::chrono::steady_clock::now() + std::chrono::milliseconds(ms);
|
embargo = std::chrono::steady_clock::now() + std::chrono::milliseconds(ms);
|
||||||
dataTransfer.enqueueItem(shared_from_this());
|
fileTransfer.enqueueItem(shared_from_this());
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
fail(exc);
|
fail(exc);
|
||||||
|
@ -456,7 +456,7 @@ struct curlDataTransfer : public DataTransfer
|
||||||
|
|
||||||
std::thread workerThread;
|
std::thread workerThread;
|
||||||
|
|
||||||
curlDataTransfer()
|
curlFileTransfer()
|
||||||
: mt19937(rd())
|
: mt19937(rd())
|
||||||
{
|
{
|
||||||
static std::once_flag globalInit;
|
static std::once_flag globalInit;
|
||||||
|
@ -469,7 +469,7 @@ struct curlDataTransfer : public DataTransfer
|
||||||
#endif
|
#endif
|
||||||
#if LIBCURL_VERSION_NUM >= 0x071e00 // Max connections requires >= 7.30.0
|
#if LIBCURL_VERSION_NUM >= 0x071e00 // Max connections requires >= 7.30.0
|
||||||
curl_multi_setopt(curlm, CURLMOPT_MAX_TOTAL_CONNECTIONS,
|
curl_multi_setopt(curlm, CURLMOPT_MAX_TOTAL_CONNECTIONS,
|
||||||
dataTransferSettings.httpConnections.get());
|
fileTransferSettings.httpConnections.get());
|
||||||
#endif
|
#endif
|
||||||
|
|
||||||
wakeupPipe.create();
|
wakeupPipe.create();
|
||||||
|
@ -478,7 +478,7 @@ struct curlDataTransfer : public DataTransfer
|
||||||
workerThread = std::thread([&]() { workerThreadEntry(); });
|
workerThread = std::thread([&]() { workerThreadEntry(); });
|
||||||
}
|
}
|
||||||
|
|
||||||
~curlDataTransfer()
|
~curlFileTransfer()
|
||||||
{
|
{
|
||||||
stopWorkerThread();
|
stopWorkerThread();
|
||||||
|
|
||||||
|
@ -641,8 +641,8 @@ struct curlDataTransfer : public DataTransfer
|
||||||
}
|
}
|
||||||
#endif
|
#endif
|
||||||
|
|
||||||
void enqueueDataTransfer(const DataTransferRequest & request,
|
void enqueueFileTransfer(const FileTransferRequest & request,
|
||||||
Callback<DataTransferResult> callback) override
|
Callback<FileTransferResult> callback) override
|
||||||
{
|
{
|
||||||
/* Ugly hack to support s3:// URIs. */
|
/* Ugly hack to support s3:// URIs. */
|
||||||
if (hasPrefix(request.uri, "s3://")) {
|
if (hasPrefix(request.uri, "s3://")) {
|
||||||
|
@ -660,9 +660,9 @@ struct curlDataTransfer : public DataTransfer
|
||||||
|
|
||||||
// FIXME: implement ETag
|
// FIXME: implement ETag
|
||||||
auto s3Res = s3Helper.getObject(bucketName, key);
|
auto s3Res = s3Helper.getObject(bucketName, key);
|
||||||
DataTransferResult res;
|
FileTransferResult res;
|
||||||
if (!s3Res.data)
|
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;
|
res.data = s3Res.data;
|
||||||
callback(std::move(res));
|
callback(std::move(res));
|
||||||
#else
|
#else
|
||||||
|
@ -676,22 +676,22 @@ struct curlDataTransfer : public DataTransfer
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
ref<DataTransfer> getDataTransfer()
|
ref<FileTransfer> getFileTransfer()
|
||||||
{
|
{
|
||||||
static ref<DataTransfer> dataTransfer = makeDataTransfer();
|
static ref<FileTransfer> fileTransfer = makeFileTransfer();
|
||||||
return dataTransfer;
|
return fileTransfer;
|
||||||
}
|
}
|
||||||
|
|
||||||
ref<DataTransfer> makeDataTransfer()
|
ref<FileTransfer> makeFileTransfer()
|
||||||
{
|
{
|
||||||
return make_ref<curlDataTransfer>();
|
return make_ref<curlFileTransfer>();
|
||||||
}
|
}
|
||||||
|
|
||||||
std::future<DataTransferResult> DataTransfer::enqueueDataTransfer(const DataTransferRequest & request)
|
std::future<FileTransferResult> FileTransfer::enqueueFileTransfer(const FileTransferRequest & request)
|
||||||
{
|
{
|
||||||
auto promise = std::make_shared<std::promise<DataTransferResult>>();
|
auto promise = std::make_shared<std::promise<FileTransferResult>>();
|
||||||
enqueueDataTransfer(request,
|
enqueueFileTransfer(request,
|
||||||
{[promise](std::future<DataTransferResult> fut) {
|
{[promise](std::future<FileTransferResult> fut) {
|
||||||
try {
|
try {
|
||||||
promise->set_value(fut.get());
|
promise->set_value(fut.get());
|
||||||
} catch (...) {
|
} catch (...) {
|
||||||
|
@ -701,21 +701,21 @@ std::future<DataTransferResult> DataTransfer::enqueueDataTransfer(const DataTran
|
||||||
return promise->get_future();
|
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 */
|
/* 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
|
/* 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
|
thread. If 'sink' is a coroutine, this will fail. Also, if the
|
||||||
sink is expensive (e.g. one that does decompression and writing
|
sink is expensive (e.g. one that does decompression and writing
|
||||||
to the Nix store), it would stall the download thread too much.
|
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();
|
state->avail.notify_one();
|
||||||
};
|
};
|
||||||
|
|
||||||
enqueueDataTransfer(request,
|
enqueueFileTransfer(request,
|
||||||
{[_state](std::future<DataTransferResult> fut) {
|
{[_state](std::future<FileTransferResult> fut) {
|
||||||
auto state(_state->lock());
|
auto state(_state->lock());
|
||||||
state->quit = true;
|
state->quit = true;
|
||||||
try {
|
try {
|
||||||
|
|
|
@ -9,7 +9,7 @@
|
||||||
|
|
||||||
namespace nix {
|
namespace nix {
|
||||||
|
|
||||||
struct DataTransferSettings : Config
|
struct FileTransferSettings : Config
|
||||||
{
|
{
|
||||||
Setting<bool> enableHttp2{this, true, "http2",
|
Setting<bool> enableHttp2{this, true, "http2",
|
||||||
"Whether to enable HTTP/2 support."};
|
"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."};
|
"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 uri;
|
||||||
std::string expectedETag;
|
std::string expectedETag;
|
||||||
bool verifyTLS = true;
|
bool verifyTLS = true;
|
||||||
bool head = false;
|
bool head = false;
|
||||||
size_t tries = dataTransferSettings.tries;
|
size_t tries = fileTransferSettings.tries;
|
||||||
unsigned int baseRetryTimeMs = 250;
|
unsigned int baseRetryTimeMs = 250;
|
||||||
ActivityId parentAct;
|
ActivityId parentAct;
|
||||||
bool decompress = true;
|
bool decompress = true;
|
||||||
|
@ -47,7 +47,7 @@ struct DataTransferRequest
|
||||||
std::string mimeType;
|
std::string mimeType;
|
||||||
std::function<void(char *, size_t)> dataCallback;
|
std::function<void(char *, size_t)> dataCallback;
|
||||||
|
|
||||||
DataTransferRequest(const std::string & uri)
|
FileTransferRequest(const std::string & uri)
|
||||||
: uri(uri), parentAct(getCurActivity()) { }
|
: uri(uri), parentAct(getCurActivity()) { }
|
||||||
|
|
||||||
std::string verb()
|
std::string verb()
|
||||||
|
@ -56,7 +56,7 @@ struct DataTransferRequest
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
struct DataTransferResult
|
struct FileTransferResult
|
||||||
{
|
{
|
||||||
bool cached = false;
|
bool cached = false;
|
||||||
std::string etag;
|
std::string etag;
|
||||||
|
@ -67,43 +67,43 @@ struct DataTransferResult
|
||||||
|
|
||||||
class Store;
|
class Store;
|
||||||
|
|
||||||
struct DataTransfer
|
struct FileTransfer
|
||||||
{
|
{
|
||||||
virtual ~DataTransfer() { }
|
virtual ~FileTransfer() { }
|
||||||
|
|
||||||
/* Enqueue a data transfer 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 DataTransferError
|
the download. The future may throw a FileTransferError
|
||||||
exception. */
|
exception. */
|
||||||
virtual void enqueueDataTransfer(const DataTransferRequest & request,
|
virtual void enqueueFileTransfer(const FileTransferRequest & request,
|
||||||
Callback<DataTransferResult> callback) = 0;
|
Callback<FileTransferResult> callback) = 0;
|
||||||
|
|
||||||
std::future<DataTransferResult> enqueueDataTransfer(const DataTransferRequest & request);
|
std::future<FileTransferResult> enqueueFileTransfer(const FileTransferRequest & request);
|
||||||
|
|
||||||
/* Synchronously download a file. */
|
/* Synchronously download a file. */
|
||||||
DataTransferResult download(const DataTransferRequest & request);
|
FileTransferResult download(const FileTransferRequest & request);
|
||||||
|
|
||||||
/* Synchronously upload a file. */
|
/* 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
|
/* Download a file, writing its data to a sink. The sink will be
|
||||||
invoked on the thread of the caller. */
|
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 };
|
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. */
|
because it enables connection reuse and HTTP/2 multiplexing. */
|
||||||
ref<DataTransfer> getDataTransfer();
|
ref<FileTransfer> getFileTransfer();
|
||||||
|
|
||||||
/* Return a new DataTransfer object. */
|
/* Return a new FileTransfer object. */
|
||||||
ref<DataTransfer> makeDataTransfer();
|
ref<FileTransfer> makeFileTransfer();
|
||||||
|
|
||||||
class DataTransferError : public Error
|
class FileTransferError : public Error
|
||||||
{
|
{
|
||||||
public:
|
public:
|
||||||
DataTransfer::Error error;
|
FileTransfer::Error error;
|
||||||
DataTransferError(DataTransfer::Error error, const FormatOrString & fs)
|
FileTransferError(FileTransfer::Error error, const FormatOrString & fs)
|
||||||
: Error(fs), error(error)
|
: Error(fs), error(error)
|
||||||
{ }
|
{ }
|
||||||
};
|
};
|
||||||
|
|
|
@ -85,14 +85,14 @@ protected:
|
||||||
checkEnabled();
|
checkEnabled();
|
||||||
|
|
||||||
try {
|
try {
|
||||||
DataTransferRequest request(cacheUri + "/" + path);
|
FileTransferRequest request(cacheUri + "/" + path);
|
||||||
request.head = true;
|
request.head = true;
|
||||||
getDataTransfer()->download(request);
|
getFileTransfer()->download(request);
|
||||||
return true;
|
return true;
|
||||||
} catch (DataTransferError & e) {
|
} catch (FileTransferError & e) {
|
||||||
/* S3 buckets return 403 if a file doesn't exist and the
|
/* S3 buckets return 403 if a file doesn't exist and the
|
||||||
bucket is unlistable, so treat 403 as 404. */
|
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;
|
return false;
|
||||||
maybeDisable();
|
maybeDisable();
|
||||||
throw;
|
throw;
|
||||||
|
@ -103,19 +103,19 @@ protected:
|
||||||
const std::string & data,
|
const std::string & data,
|
||||||
const std::string & mimeType) override
|
const std::string & mimeType) override
|
||||||
{
|
{
|
||||||
auto req = DataTransferRequest(cacheUri + "/" + path);
|
auto req = FileTransferRequest(cacheUri + "/" + path);
|
||||||
req.data = std::make_shared<string>(data); // FIXME: inefficient
|
req.data = std::make_shared<string>(data); // FIXME: inefficient
|
||||||
req.mimeType = mimeType;
|
req.mimeType = mimeType;
|
||||||
try {
|
try {
|
||||||
getDataTransfer()->upload(req);
|
getFileTransfer()->upload(req);
|
||||||
} catch (DataTransferError & e) {
|
} catch (FileTransferError & e) {
|
||||||
throw UploadToHTTP("while uploading to HTTP binary cache at '%s': %s", cacheUri, e.msg());
|
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;
|
return request;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -124,9 +124,9 @@ protected:
|
||||||
checkEnabled();
|
checkEnabled();
|
||||||
auto request(makeRequest(path));
|
auto request(makeRequest(path));
|
||||||
try {
|
try {
|
||||||
getDataTransfer()->download(std::move(request), sink);
|
getFileTransfer()->download(std::move(request), sink);
|
||||||
} catch (DataTransferError & e) {
|
} catch (FileTransferError & e) {
|
||||||
if (e.error == DataTransfer::NotFound || e.error == DataTransfer::Forbidden)
|
if (e.error == FileTransfer::NotFound || e.error == FileTransfer::Forbidden)
|
||||||
throw NoSuchBinaryCacheFile("file '%s' does not exist in binary cache '%s'", path, getUri());
|
throw NoSuchBinaryCacheFile("file '%s' does not exist in binary cache '%s'", path, getUri());
|
||||||
maybeDisable();
|
maybeDisable();
|
||||||
throw;
|
throw;
|
||||||
|
@ -142,12 +142,12 @@ protected:
|
||||||
|
|
||||||
auto callbackPtr = std::make_shared<decltype(callback)>(std::move(callback));
|
auto callbackPtr = std::make_shared<decltype(callback)>(std::move(callback));
|
||||||
|
|
||||||
getDataTransfer()->enqueueDataTransfer(request,
|
getFileTransfer()->enqueueFileTransfer(request,
|
||||||
{[callbackPtr, this](std::future<DataTransferResult> result) {
|
{[callbackPtr, this](std::future<FileTransferResult> result) {
|
||||||
try {
|
try {
|
||||||
(*callbackPtr)(result.get().data);
|
(*callbackPtr)(result.get().data);
|
||||||
} catch (DataTransferError & e) {
|
} catch (FileTransferError & e) {
|
||||||
if (e.error == DataTransfer::NotFound || e.error == DataTransfer::Forbidden)
|
if (e.error == FileTransfer::NotFound || e.error == FileTransfer::Forbidden)
|
||||||
return (*callbackPtr)(std::shared_ptr<std::string>());
|
return (*callbackPtr)(std::shared_ptr<std::string>());
|
||||||
maybeDisable();
|
maybeDisable();
|
||||||
callbackPtr->rethrow();
|
callbackPtr->rethrow();
|
||||||
|
|
|
@ -132,7 +132,7 @@ ref<Aws::Client::ClientConfiguration> S3Helper::makeConfig(const string & region
|
||||||
return res;
|
return res;
|
||||||
}
|
}
|
||||||
|
|
||||||
S3Helper::DataTransferResult S3Helper::getObject(
|
S3Helper::FileTransferResult S3Helper::getObject(
|
||||||
const std::string & bucketName, const std::string & key)
|
const std::string & bucketName, const std::string & key)
|
||||||
{
|
{
|
||||||
debug("fetching 's3://%s/%s'...", bucketName, key);
|
debug("fetching 's3://%s/%s'...", bucketName, key);
|
||||||
|
@ -146,7 +146,7 @@ S3Helper::DataTransferResult S3Helper::getObject(
|
||||||
return Aws::New<std::stringstream>("STRINGSTREAM");
|
return Aws::New<std::stringstream>("STRINGSTREAM");
|
||||||
});
|
});
|
||||||
|
|
||||||
DataTransferResult res;
|
FileTransferResult res;
|
||||||
|
|
||||||
auto now1 = std::chrono::steady_clock::now();
|
auto now1 = std::chrono::steady_clock::now();
|
||||||
|
|
||||||
|
|
|
@ -18,13 +18,13 @@ struct S3Helper
|
||||||
|
|
||||||
ref<Aws::Client::ClientConfiguration> makeConfig(const std::string & region, const std::string & scheme, const std::string & endpoint);
|
ref<Aws::Client::ClientConfiguration> makeConfig(const std::string & region, const std::string & scheme, const std::string & endpoint);
|
||||||
|
|
||||||
struct DataTransferResult
|
struct FileTransferResult
|
||||||
{
|
{
|
||||||
std::shared_ptr<std::string> data;
|
std::shared_ptr<std::string> data;
|
||||||
unsigned int durationMs;
|
unsigned int durationMs;
|
||||||
};
|
};
|
||||||
|
|
||||||
DataTransferResult getObject(
|
FileTransferResult getObject(
|
||||||
const std::string & bucketName, const std::string & key);
|
const std::string & bucketName, const std::string & key);
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
|
@ -198,7 +198,7 @@ bool handleJSONLogMessage(const std::string & msg,
|
||||||
|
|
||||||
if (action == "start") {
|
if (action == "start") {
|
||||||
auto type = (ActivityType) json["type"];
|
auto type = (ActivityType) json["type"];
|
||||||
if (trusted || type == actDataTransfer)
|
if (trusted || type == actFileTransfer)
|
||||||
activities.emplace(std::piecewise_construct,
|
activities.emplace(std::piecewise_construct,
|
||||||
std::forward_as_tuple(json["id"]),
|
std::forward_as_tuple(json["id"]),
|
||||||
std::forward_as_tuple(*logger, (Verbosity) json["level"], type,
|
std::forward_as_tuple(*logger, (Verbosity) json["level"], type,
|
||||||
|
|
|
@ -17,7 +17,7 @@ typedef enum {
|
||||||
typedef enum {
|
typedef enum {
|
||||||
actUnknown = 0,
|
actUnknown = 0,
|
||||||
actCopyPath = 100,
|
actCopyPath = 100,
|
||||||
actDataTransfer = 101,
|
actFileTransfer = 101,
|
||||||
actRealise = 102,
|
actRealise = 102,
|
||||||
actCopyPaths = 103,
|
actCopyPaths = 103,
|
||||||
actBuilds = 104,
|
actBuilds = 104,
|
||||||
|
|
|
@ -180,9 +180,9 @@ static int _main(int argc, char * * argv)
|
||||||
|
|
||||||
FdSink sink(fd.get());
|
FdSink sink(fd.get());
|
||||||
|
|
||||||
DataTransferRequest req(actualUri);
|
FileTransferRequest req(actualUri);
|
||||||
req.decompress = false;
|
req.decompress = false;
|
||||||
getDataTransfer()->download(std::move(req), sink);
|
getFileTransfer()->download(std::move(req), sink);
|
||||||
}
|
}
|
||||||
|
|
||||||
/* Optionally unpack the file. */
|
/* Optionally unpack the file. */
|
||||||
|
|
|
@ -176,10 +176,10 @@ void mainWrapped(int argc, char * * argv)
|
||||||
settings.useSubstitutes = false;
|
settings.useSubstitutes = false;
|
||||||
if (!settings.tarballTtl.overriden)
|
if (!settings.tarballTtl.overriden)
|
||||||
settings.tarballTtl = std::numeric_limits<unsigned int>::max();
|
settings.tarballTtl = std::numeric_limits<unsigned int>::max();
|
||||||
if (!dataTransferSettings.tries.overriden)
|
if (!fileTransferSettings.tries.overriden)
|
||||||
dataTransferSettings.tries = 0;
|
fileTransferSettings.tries = 0;
|
||||||
if (!dataTransferSettings.connectTimeout.overriden)
|
if (!fileTransferSettings.connectTimeout.overriden)
|
||||||
dataTransferSettings.connectTimeout = 1;
|
fileTransferSettings.connectTimeout = 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (args.refresh)
|
if (args.refresh)
|
||||||
|
|
|
@ -190,8 +190,8 @@ public:
|
||||||
i->s = fmt("querying " ANSI_BOLD "%s" ANSI_NORMAL " on %s", name, getS(fields, 1));
|
i->s = fmt("querying " ANSI_BOLD "%s" ANSI_NORMAL " on %s", name, getS(fields, 1));
|
||||||
}
|
}
|
||||||
|
|
||||||
if ((type == actDataTransfer && hasAncestor(*state, actCopyPath, parent))
|
if ((type == actFileTransfer && hasAncestor(*state, actCopyPath, parent))
|
||||||
|| (type == actDataTransfer && hasAncestor(*state, actQueryPathInfo, parent))
|
|| (type == actFileTransfer && hasAncestor(*state, actQueryPathInfo, parent))
|
||||||
|| (type == actCopyPath && hasAncestor(*state, actSubstitute, parent)))
|
|| (type == actCopyPath && hasAncestor(*state, actSubstitute, parent)))
|
||||||
i->visible = false;
|
i->visible = false;
|
||||||
|
|
||||||
|
@ -416,7 +416,7 @@ public:
|
||||||
if (!s2.empty()) { res += " ("; res += s2; res += ')'; }
|
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");
|
auto s = renderActivity(actOptimiseStore, "%s paths optimised");
|
||||||
|
|
|
@ -138,8 +138,8 @@ struct CmdUpgradeNix : MixDryRun, StoreCommand
|
||||||
Activity act(*logger, lvlInfo, actUnknown, "querying latest Nix version");
|
Activity act(*logger, lvlInfo, actUnknown, "querying latest Nix version");
|
||||||
|
|
||||||
// FIXME: use nixos.org?
|
// FIXME: use nixos.org?
|
||||||
auto req = DataTransferRequest(storePathsUrl);
|
auto req = FileTransferRequest(storePathsUrl);
|
||||||
auto res = getDataTransfer()->download(req);
|
auto res = getFileTransfer()->download(req);
|
||||||
|
|
||||||
auto state = std::make_unique<EvalState>(Strings(), store);
|
auto state = std::make_unique<EvalState>(Strings(), store);
|
||||||
auto v = state->allocValue();
|
auto v = state->allocValue();
|
||||||
|
|
Loading…
Reference in a new issue