guiwf/lix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

libstore: Fix libcurl thread wakeup with curl >= 8.21

Browse Source 
Since https://github.com/curl/curl/commit/2a2104f3cff44bb28bb570a093be52bbeeed8f23
libcurl sometimes just swallows events in curl_multi_perform, we entirely miss them,
and curl_multi_poll blocks forever.

Fixes #1234.

Based on https://github.com/NixOS/nix/pull/16090.

Co-authored-by: Sergei Zimmerman <sergei@zimmerman.foo>
Change-Id: I453dce192de0ea9e92bfb0114020dc9d6a6a6964
This commit is contained in:
Qyriad
2026-07-09 15:09:26 +00:00
co-authored by Sergei Zimmerman
parent 41bad096e3
commit 81084c7fc9
3 changed files with 76 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/libstore/curlmulti.cc
+70 -62
View File
@@ -23,6 +23,30 @@
namespace nix {
void CurlMulti::State::quit()
{
quitting = true;
// We will not be processing any more requests.
for (auto & item : incoming) {
item->finish(CURLE_ABORTED_BY_CALLBACK);
}
incoming.clear();
unpause.clear();
// make a note that we're dying and acknowledge all pending cancel
// requests by individual transfers. not doing this can cause bugs
// like #1218 in which the process deadlocks waiting for transfers
// to cancel with no download thread to make this happen; this was
// likely caused by a transfer requesting a cancellation *exactly*
// before a signal was received, causing the curl thread to die in
// a hurry without processing cancellations. the transfer is stuck
// from that point on, and since this happened in a destructor the
// entire process locked up solid. curl exceptions could have also
// caused this; we set the `quit` flag just in case to avoid this.
for (auto & [item, promise] : cancel) {
promise.set_value();
}
}
CurlMulti::CurlMulti(unsigned int baseRetryTimeMs)
: curlm(curl_multi_init())
, baseRetryTimeMs(baseRetryTimeMs)
@@ -64,7 +88,7 @@ void CurlMulti::unpause(const std::shared_ptr<TransferItem> & transfer)
{
auto lock = state_.lock();
lock->unpause.push_back(transfer);
wakeup();
wakeup(*lock);
}
void CurlMulti::cancel(const std::shared_ptr<TransferItem> & transfer)
@@ -73,17 +97,18 @@ void CurlMulti::cancel(const std::shared_ptr<TransferItem> & transfer)
auto wait = promise.get_future();
{
auto lock = state_.lock();
if (lock->quit) {
if (lock->quitting) {
return;
}
lock->cancel[transfer] = std::move(promise);
wakeup(*lock);
}
wakeup();
wait.get();
}
void CurlMulti::wakeup()
void CurlMulti::wakeup(State & locked)
{
locked.workAvailable = true;
if (auto mc = curl_multi_wakeup(curlm.get()))
throw nix::Error("unexpected error from curl_multi_wakeup(): %s", curl_multi_strerror(mc));
}
@@ -91,11 +116,9 @@ void CurlMulti::wakeup()
void CurlMulti::stopWorkerThread()
{
/* Signal the worker thread to exit. */
{
auto state(state_.lock());
state->quit = true;
}
wakeup();
auto state(state_.lock());
state->quit();
wakeup(*state);
}
void CurlMulti::workerThreadMain()
@@ -116,21 +139,8 @@ void CurlMulti::workerThreadMain()
item->finish(CURLE_ABORTED_BY_CALLBACK);
}
// make a note that we're dying and acknowledge all pending cancel
// requests by individual transfers. not doing this can cause bugs
// like #1218 in which the process deadlocks waiting for transfers
// to cancel with no download thread to make this happen; this was
// likely caused by a transfer requesting a cancellation *exactly*
// before a signal was received, causing the curl thread to die in
// a hurry without processing cancellations. the transfer is stuck
// from that point on, and since this happened in a destructor the
// entire process locked up solid. curl exceptions could have also
// caused this; we set the `quit` flag just in case to avoid this.
auto lock = state_.lock();
lock->quit = true;
for (auto & [item, promise] : lock->cancel) {
promise.set_value();
}
lock->quit();
});
bool quit = false;
@@ -139,16 +149,7 @@ void CurlMulti::workerThreadMain()
// loop with kj. until then curl will handle its timeouts internally.
int64_t timeoutMs = INT64_MAX;
while (true) {
{
auto cancel = [&] { return std::exchange(state_.lock()->cancel, {}); }();
for (auto & [item, promise] : cancel) {
curl_multi_remove_handle(curlm.get(), item->req.get());
items.erase(item->req.get());
promise.set_value();
}
}
while (!quit) {
/* Let curl do its thing. */
int running;
CURLMcode mc = curl_multi_perform(curlm.get(), &running);
@@ -168,37 +169,31 @@ void CurlMulti::workerThreadMain()
}
}
// exit immediately and abort all running transfers. waiting for transfers to finish
// before exiting this loop may hang the shutdown procedure forever, e.g. if blocked
// transfers would be destroyed (thus aborted) after the curl thread for any reason.
if (quit) {
break;
{
auto cancel = [&] { return std::exchange(state_.lock()->cancel, {}); }();
for (auto & [item, promise] : cancel) {
curl_multi_remove_handle(curlm.get(), item->req.get());
items.erase(item->req.get());
promise.set_value();
}
}
/* Wait for activity, including wakeup events. */
mc = curl_multi_poll(curlm.get(), nullptr, 0, std::min<int64_t>(timeoutMs, INT_MAX), nullptr);
if (mc != CURLM_OK)
throw nix::Error("unexpected error from curl_multi_poll(): %s", curl_multi_strerror(mc));
/* Add new curl requests from the incoming requests queue,
except for requests that are embargoed (waiting for a
retry timeout to expire). */
std::vector<std::shared_ptr<TransferItem>> incoming;
std::vector<std::shared_ptr<TransferItem>> unpause;
bool haveWork = false;
timeoutMs = INT64_MAX;
{
auto unpause = [&] { return std::exchange(state_.lock()->unpause, {}); }();
for (auto & item : unpause) {
curl_easy_pause(item->req.get(), CURLPAUSE_CONT);
}
}
timeoutMs = 10'000;
{
auto state(state_.lock());
incoming = std::exchange(state->incoming, {});
quit = state->quit;
unpause = std::exchange(state->unpause, {});
haveWork = std::exchange(state->workAvailable, false);
quit = state->quitting;
}
for (auto & item : incoming) {
@@ -206,6 +201,26 @@ void CurlMulti::workerThreadMain()
curl_multi_add_handle(curlm.get(), item->req.get());
items[item->req.get()] = item;
}
// exit immediately and abort all running transfers. waiting for transfers to finish
// before exiting this loop may hang the shutdown procedure forever, e.g. if blocked
// transfers would be destroyed (thus aborted) after the curl thread for any reason.
if (quit) {
break;
}
for (auto & item : unpause) {
curl_easy_pause(item->req.get(), CURLPAUSE_CONT);
}
if (haveWork) {
timeoutMs = 0;
}
/* Wait for activity, including wakeup events. */
mc = curl_multi_poll(curlm.get(), nullptr, 0, timeoutMs, nullptr);
if (mc != CURLM_OK)
throw nix::Error("unexpected error from curl_multi_poll(): %s", curl_multi_strerror(mc));
}
debug("download thread shutting down");
@@ -222,14 +237,7 @@ void CurlMulti::workerThreadEntry()
printError("unexpected error in download thread");
}
{
auto state(state_.lock());
for (auto & item : state->incoming) {
item->finish(CURLE_ABORTED_BY_CALLBACK);
}
state->incoming.clear();
state->quit = true;
}
state_.lock()->quit();
}
void CurlMulti::enqueueItem(std::shared_ptr<TransferItem> item)
@@ -241,11 +249,11 @@ void CurlMulti::enqueueItem(std::shared_ptr<TransferItem> item)
{
auto state(state_.lock());
if (state->quit)
if (state->quitting)
throw nix::Error("cannot enqueue download request because the download thread is shutting down");
state->incoming.push_back(item);
wakeup(*state);
}
wakeup();
}
} // namespace nix
lix/libstore/curlmulti.hh
+5 -2
View File
@@ -20,10 +20,13 @@ struct CurlMulti
public:
struct State
{
bool quit = false;
bool quitting = false;
bool workAvailable = false;
std::vector<std::shared_ptr<TransferItem>> incoming;
std::vector<std::shared_ptr<TransferItem>> unpause;
std::map<std::shared_ptr<TransferItem>, std::promise<void>> cancel;
void quit();
};
// Fields.
@@ -44,7 +47,7 @@ public:
void cancel(std::shared_ptr<TransferItem> const & transfer);
void wakeup();
void wakeup(State & locked);
void stopWorkerThread();
lix/libstore/filetransfer.cc
+1 -1
View File
@@ -30,7 +30,7 @@ ref<FileTransfer> getFileTransfer()
{
static ref<curlFileTransfer> fileTransfer = makeCurlFileTransfer({});
if (fileTransfer->multi->state_.lock()->quit) {
if (fileTransfer->multi->state_.lock()->quitting) {
fileTransfer = makeCurlFileTransfer({});
}