diff --git a/lix/libstore/curlmulti.cc b/lix/libstore/curlmulti.cc index 1399e6ae6..628c72445 100644 --- a/lix/libstore/curlmulti.cc +++ b/lix/libstore/curlmulti.cc @@ -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 & transfer) { auto lock = state_.lock(); lock->unpause.push_back(transfer); - wakeup(); + wakeup(*lock); } void CurlMulti::cancel(const std::shared_ptr & transfer) @@ -73,17 +97,18 @@ void CurlMulti::cancel(const std::shared_ptr & 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(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> incoming; + std::vector> 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 item) @@ -241,11 +249,11 @@ void CurlMulti::enqueueItem(std::shared_ptr 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 diff --git a/lix/libstore/curlmulti.hh b/lix/libstore/curlmulti.hh index 312b2fca4..6de71f75e 100644 --- a/lix/libstore/curlmulti.hh +++ b/lix/libstore/curlmulti.hh @@ -20,10 +20,13 @@ struct CurlMulti public: struct State { - bool quit = false; + bool quitting = false; + bool workAvailable = false; std::vector> incoming; std::vector> unpause; std::map, std::promise> cancel; + + void quit(); }; // Fields. @@ -44,7 +47,7 @@ public: void cancel(std::shared_ptr const & transfer); - void wakeup(); + void wakeup(State & locked); void stopWorkerThread(); diff --git a/lix/libstore/filetransfer.cc b/lix/libstore/filetransfer.cc index 6457f0d0d..79238c1f4 100644 --- a/lix/libstore/filetransfer.cc +++ b/lix/libstore/filetransfer.cc @@ -30,7 +30,7 @@ ref getFileTransfer() { static ref fileTransfer = makeCurlFileTransfer({}); - if (fileTransfer->multi->state_.lock()->quit) { + if (fileTransfer->multi->state_.lock()->quitting) { fileTransfer = makeCurlFileTransfer({}); }