From e47bff547ac00b0ca12f2d370c0538397f8c3e6f Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Fri, 21 Feb 2025 00:35:11 +0100 Subject: [PATCH 1/3] libutil: fix some thread pool bugs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 4138fc7622bcd23613050073864076591b76d519 mistakenly removed an early exit from non-main worker threads. this led to exceptions not ending `process()` calls in a timely manner and instead draining the entire work queue first, which for e.g. Interrupted errors would cause many duplicated reports per worker thread instead of only one per thread. it also did not properly rethrow a work item exception in all cases, e.g. when all work had completed by the time `process()` was called. due to a mistake in the thread starting check it was possible that a system would require n² work items to start n worker threads if some work items process quickly enough while other items block for a bit. Change-Id: I7d59483580cda1c19980f9296074216a0fbf2c4e --- lix/libutil/thread-pool.cc | 16 +++-- tests/unit/libutil/thread-pool.cc | 111 ++++++++++++++++++++++++++++++ tests/unit/meson.build | 1 + 3 files changed, 122 insertions(+), 6 deletions(-) create mode 100644 tests/unit/libutil/thread-pool.cc diff --git a/lix/libutil/thread-pool.cc b/lix/libutil/thread-pool.cc index 9b6d6066f..42b0591a2 100644 --- a/lix/libutil/thread-pool.cc +++ b/lix/libutil/thread-pool.cc @@ -47,23 +47,25 @@ void ThreadPool::enqueueWithAio(const work_t & t) if (quit) throw ThreadPoolShutDown("cannot enqueue a work item while the thread pool is shutting down"); state->pending.push(t); - if (state->pending.size() > state->workers.size() && state->workers.size() < maxThreads) + if (state->active == state->workers.size() && state->workers.size() < maxThreads) state->workers.emplace_back(&ThreadPool::doWork, this); work.notify_one(); } void ThreadPool::process() { - state_.lock()->draining = true; + const auto shouldWait = [&] { + auto state(state_.lock()); + state->draining = true; + return state->active > 0 || !state->pending.empty(); + }(); /* Wait until no more work is pending or active. */ try { - if (auto state(state_.lock()); state->active == 0 && state->pending.empty()) { - return; + if (shouldWait) { + quit.wait(false); } - quit.wait(false); - auto state(state_.lock()); if (state->exception) std::rethrow_exception(state->exception); @@ -141,6 +143,8 @@ void ThreadPool::doWork() /* Wait until a work item is available or we're asked to quit. */ while (true) { + if (quit) return; + if (!state->pending.empty()) break; /* If there are no active or pending items, and the diff --git a/tests/unit/libutil/thread-pool.cc b/tests/unit/libutil/thread-pool.cc new file mode 100644 index 000000000..8936548a5 --- /dev/null +++ b/tests/unit/libutil/thread-pool.cc @@ -0,0 +1,111 @@ +#include "lix/libutil/thread-pool.hh" +#include +#include +#include +#include + +static auto onThreadExit(auto fn) +{ + auto deferred = kj::defer(std::move(fn)); + return std::make_shared(std::move(deferred)); +} + +namespace nix { + +TEST(ThreadPool, creates_threads) +{ + ThreadPool t{"test", 2}; + + std::atomic_bool unblockA{false}, unblockB{false}; + std::atomic_bool started{false}; + + t.enqueue([&] { + started = true; + started.notify_all(); + unblockA.wait(false); + unblockB = true; + unblockB.notify_all(); + }); + started.wait(false); + + // now no work is pending. the next enqueue should start a + // new thread; if it does not we'll deadlock and time out. + + started = false; + t.enqueue([&] { + started = true; + started.notify_all(); + unblockB.wait(false); + }); + started.wait(false); + + unblockA = true; + unblockA.notify_all(); + + t.process(); +} + +TEST(ThreadPool, early_quit) +{ + ThreadPool t{"test", 2}; + bool ran_anyway = false; + + struct Dead : std::exception {}; + + std::atomic_bool unblockA{false}, unblockB{false}; + std::atomic_bool started{false}; + + t.enqueue([&] { + started = true; + started.notify_all(); + unblockA.wait(false); + thread_local auto _ = onThreadExit([&] { + unblockB = true; + unblockB.notify_all(); + }); + throw Dead{}; + }); + started.wait(false); + + started = false; + t.enqueue([&] { + started = true; + started.notify_all(); + unblockB.wait(false); + }); + started.wait(false); + + // this one should never run. the first thread saw an exception, + // and the second thread should have exited early because of it. + t.enqueue([&] { ran_anyway = true; }); + + unblockA = true; + unblockA.notify_all(); + + ASSERT_THROW(t.process(), Dead); + ASSERT_FALSE(ran_anyway); +} + +TEST(ThreadPool, always_rethrows) +{ + ThreadPool t{"test"}; + + struct Dead : std::exception {}; + + std::atomic_bool flag{false}; + + t.enqueue([&] { + thread_local auto _ = onThreadExit([&] { + flag = true; + flag.notify_all(); + }); + + throw Dead{}; + }); + + flag.wait(false); + + ASSERT_THROW(t.process(), Dead); +} + +} diff --git a/tests/unit/meson.build b/tests/unit/meson.build index 00a4eae42..d82f12bf7 100644 --- a/tests/unit/meson.build +++ b/tests/unit/meson.build @@ -64,6 +64,7 @@ libutil_tests_sources = files( 'libutil/serialise.cc', 'libutil/suggestions.cc', 'libutil/tests.cc', + 'libutil/thread-pool.cc', 'libutil/url.cc', 'libutil/url-name.cc', 'libutil/xml-writer.cc', From 3644b519a9da1d0ea8e49557cf7e1e0de5d5f118 Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Fri, 21 Feb 2025 00:35:11 +0100 Subject: [PATCH 2/3] libstore: await promises instead of discarding oops. Change-Id: I9f47875085500fa392f74ee51293f9a95c455997 --- lix/libstore/local-store.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lix/libstore/local-store.cc b/lix/libstore/local-store.cc index 77594cf05..4617649a2 100644 --- a/lix/libstore/local-store.cc +++ b/lix/libstore/local-store.cc @@ -1671,7 +1671,7 @@ try { StorePathSet referrers; queryReferrers(path, referrers); for (auto & i : referrers) if (i != path) { - verifyPath(i, storePathsInStoreDir, done, validPaths, repair, errors); + TRY_AWAIT(verifyPath(i, storePathsInStoreDir, done, validPaths, repair, errors)); if (validPaths.count(i)) canInvalidate = false; } From 0eb1164a8de94acd98c7024e1f285a92872409e4 Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Fri, 21 Feb 2025 00:35:11 +0100 Subject: [PATCH 3/3] deps: backport capnproto promise nodiscard PR this does not change the API exposed by the library, so it'll be safe for users to substitute stock capnp. we do want it for our own builds though, at least until we can switch to capnp 2 or fully go for rust. Change-Id: Ia8ac6af0165f6f61ee02345179dc0a93a0889a06 --- misc/capnproto-promise-nodiscard.patch | 206 +++++++++++++++++++++++++ misc/capnproto.nix | 5 + 2 files changed, 211 insertions(+) create mode 100644 misc/capnproto-promise-nodiscard.patch diff --git a/misc/capnproto-promise-nodiscard.patch b/misc/capnproto-promise-nodiscard.patch new file mode 100644 index 000000000..f4606bdc1 --- /dev/null +++ b/misc/capnproto-promise-nodiscard.patch @@ -0,0 +1,206 @@ +diff --git a/c++/WORKSPACE b/c++/WORKSPACE +index d94a279e..4871ead7 100644 +--- a/c++/WORKSPACE ++++ b/c++/WORKSPACE +@@ -31,6 +31,7 @@ cc_library( + name = "zlib", + srcs = glob(["*.c"]), + hdrs = glob(["*.h"]), ++ includes = ["."], + # Temporary workaround for zlib warnings and mac compilation, should no longer be needed with next release https://github.com/madler/zlib/issues/633 + copts = [ + "-w", +diff --git a/c++/src/kj/array.h b/c++/src/kj/array.h +index 3932f9f4..677c691a 100644 +--- a/c++/src/kj/array.h ++++ b/c++/src/kj/array.h +@@ -780,7 +780,7 @@ struct CopyConstructArray_ { + + static T* apply(T* __restrict__ pos, Iterator start, Iterator end) { + // Verify that T can be *implicitly* constructed from the source values. +- if (false) implicitCast(kj::mv(*start)); ++ if (false) (void)implicitCast(kj::mv(*start)); + + if (noexcept(T(kj::mv(*start)))) { + while (start != end) { +diff --git a/c++/src/kj/async-coroutine-test.c++ b/c++/src/kj/async-coroutine-test.c++ +index de767eca..d6ed1359 100644 +--- a/c++/src/kj/async-coroutine-test.c++ ++++ b/c++/src/kj/async-coroutine-test.c++ +@@ -288,5 +288,5 @@ KJ_TEST("Exceptions during suspended coroutine frame-unwind propagate via destru + WaitScope waitScope(loop); + + auto exception = KJ_ASSERT_NONNULL(kj::runCatchingExceptions([&]() { +- deferredThrowCoroutine(kj::NEVER_DONE); ++ (void)deferredThrowCoroutine(kj::NEVER_DONE); + })); + + KJ_EXPECT(exception.getDescription() == "thrown during unwind"); +diff --git a/c++/src/kj/async-io-test.c++ b/c++/src/kj/async-io-test.c++ +index e8892b79..dffcbb26 100644 +--- a/c++/src/kj/async-io-test.c++ ++++ b/c++/src/kj/async-io-test.c++ +@@ -1577,7 +1577,7 @@ KJ_TEST("Userland pipe pump into zero-limited pipe, no data to pump") { + auto pipe2 = newOneWayPipe(uint64_t(0)); + auto pumpPromise = KJ_ASSERT_NONNULL(pipe2.out->tryPumpFrom(*pipe.in)); + +- expectRead(*pipe2.in, ""); ++ expectRead(*pipe2.in, "").wait(ws); + pipe.out = nullptr; + KJ_EXPECT(pumpPromise.wait(ws) == 0); + } +@@ -1590,7 +1590,7 @@ KJ_TEST("Userland pipe pump into zero-limited pipe, data is pumped") { + auto pipe2 = newOneWayPipe(uint64_t(0)); + auto pumpPromise = KJ_ASSERT_NONNULL(pipe2.out->tryPumpFrom(*pipe.in)); + +- expectRead(*pipe2.in, ""); ++ expectRead(*pipe2.in, "").wait(ws); + auto writePromise = pipe.out->write("foo", 3); + KJ_EXPECT_THROW_RECOVERABLE_MESSAGE("abortRead() has been called", pumpPromise.wait(ws)); + } +diff --git a/c++/src/kj/async.h b/c++/src/kj/async.h +index 564b5171..d4f2d55c 100644 +--- a/c++/src/kj/async.h ++++ b/c++/src/kj/async.h +@@ -118,7 +118,7 @@ private: + // Promises + + template +-class Promise: protected _::PromiseBase { ++class [[nodiscard]] Promise: protected _::PromiseBase { + // The basic primitive of asynchronous computation in KJ. Similar to "futures", but designed + // specifically for event loop concurrency. Similar to E promises and JavaScript Promises/A. + // +diff --git a/c++/src/kj/common-test.c++ b/c++/src/kj/common-test.c++ +index 97856125..913a6be4 100644 +--- a/c++/src/kj/common-test.c++ ++++ b/c++/src/kj/common-test.c++ +@@ -573,7 +573,7 @@ TEST(Common, Downcast) { + + EXPECT_EQ(&bar, &downcast(foo)); + #if defined(KJ_DEBUG) && !KJ_NO_RTTI +- KJ_EXPECT_THROW_MESSAGE("Value cannot be downcast", downcast(foo)); ++ KJ_EXPECT_THROW_MESSAGE("Value cannot be downcast", (void)downcast(foo)); + #endif + + #if KJ_NO_RTTI +diff --git a/c++/src/kj/compat/http-test.c++ b/c++/src/kj/compat/http-test.c++ +index f10ff8d1..9003099d 100644 +--- a/c++/src/kj/compat/http-test.c++ ++++ b/c++/src/kj/compat/http-test.c++ +@@ -6553,7 +6553,7 @@ KJ_TEST("Simple CONNECT Server works") { + "\r\n" + "hello"_kj).wait(waitScope); + +- expectEnd(*pipe.ends[1]); ++ expectEnd(*pipe.ends[1]).wait(waitScope); + + listenTask.wait(waitScope); + +@@ -6628,7 +6628,7 @@ KJ_TEST("CONNECT Server (201 status)") { + "\r\n" + "hello"_kj).wait(waitScope); + +- expectEnd(*pipe.ends[1]); ++ expectEnd(*pipe.ends[1]).wait(waitScope); + + listenTask.wait(waitScope); + +@@ -6706,7 +6706,7 @@ KJ_TEST("CONNECT Server rejected") { + "\r\n" + "boom"_kj).wait(waitScope); + +- expectEnd(*pipe.ends[1]); ++ expectEnd(*pipe.ends[1]).wait(waitScope); + + listenTask.wait(waitScope); + +@@ -6774,7 +6774,7 @@ KJ_TEST("CONNECT Server cancels read") { + "HTTP/1.1 200 OK\r\n" + "\r\n"_kj).wait(waitScope); + +- expectEnd(*pipe.ends[1]); ++ expectEnd(*pipe.ends[1]).wait(waitScope); + + listenTask.wait(waitScope); + } +@@ -6840,7 +6840,7 @@ KJ_TEST("CONNECT Server cancels write") { + "HTTP/1.1 200 OK\r\n" + "\r\n"_kj).wait(waitScope); + +- expectEnd(*pipe.ends[1]); ++ expectEnd(*pipe.ends[1]).wait(waitScope); + + listenTask.wait(waitScope); + } +@@ -6913,7 +6913,7 @@ KJ_TEST("CONNECT rejects Transfer-Encoding") { + "\r\n" + "ERROR: Bad Request"_kj).wait(waitScope); + +- expectEnd(*pipe.ends[1]); ++ expectEnd(*pipe.ends[1]).wait(waitScope); + + listenTask.wait(waitScope); + } +@@ -6947,7 +6947,7 @@ KJ_TEST("CONNECT rejects Content-Length") { + "\r\n" + "ERROR: Bad Request"_kj).wait(waitScope); + +- expectEnd(*pipe.ends[1]); ++ expectEnd(*pipe.ends[1]).wait(waitScope); + + listenTask.wait(waitScope); + } +diff --git a/c++/src/kj/compat/tls-test.c++ b/c++/src/kj/compat/tls-test.c++ +index dddefa57..52ccc68a 100644 +--- a/c++/src/kj/compat/tls-test.c++ ++++ b/c++/src/kj/compat/tls-test.c++ +@@ -1037,15 +1037,15 @@ KJ_TEST("TLS receiver experiences pre-TLS error") { + TlsReceiverTest test; + + KJ_LOG(INFO, "Accepting before a bad connect"); +- auto promise = test.receiver->accept(); ++ auto acceptPromise = test.receiver->accept(); + + KJ_LOG(INFO, "Disappointing our server"); +- test.baseReceiver->badConnect(); ++ auto connectPromise = test.baseReceiver->badConnect(); + + // Can't use KJ_EXPECT_THROW_RECOVERABLE_MESSAGE because wait() that returns a value can't throw + // recoverable exceptions. Can't use KJ_EXPECT_THROW_MESSAGE because non-recoverable exceptions + // will fork() in -fno-exception which screws up our state. +- promise.then([](auto) { ++ acceptPromise.then([](auto) { + KJ_FAIL_EXPECT("expected exception"); + }, [](kj::Exception&& e) { + KJ_EXPECT(e.getDescription() == "Pipes are leaky"); +diff --git a/c++/src/kj/test.h b/c++/src/kj/test.h +index 5acbb00d..de5efec2 100644 +--- a/c++/src/kj/test.h ++++ b/c++/src/kj/test.h +@@ -92,6 +92,7 @@ private: + else KJ_FAIL_EXPECT("failed: expected " #cond, _kjCondition, ##__VA_ARGS__) + #endif + ++// TODO(msvc): cast results to void like non-MSVC versions do + #if _MSC_VER && !defined(__clang__) + #define KJ_EXPECT_THROW_RECOVERABLE(type, code, ...) \ + do { \ +@@ -115,7 +116,7 @@ private: + #else + #define KJ_EXPECT_THROW_RECOVERABLE(type, code, ...) \ + do { \ +- KJ_IF_MAYBE(e, ::kj::runCatchingExceptions([&]() { code; })) { \ ++ KJ_IF_MAYBE(e, ::kj::runCatchingExceptions([&]() { (void)({code}); })) { \ + KJ_EXPECT(e->getType() == ::kj::Exception::Type::type, \ + "code threw wrong exception type: " #code, *e, ##__VA_ARGS__); \ + } else { \ +@@ -125,7 +126,7 @@ private: + + #define KJ_EXPECT_THROW_RECOVERABLE_MESSAGE(message, code, ...) \ + do { \ +- KJ_IF_MAYBE(e, ::kj::runCatchingExceptions([&]() { code; })) { \ ++ KJ_IF_MAYBE(e, ::kj::runCatchingExceptions([&]() { (void)({code}); })) { \ + KJ_EXPECT(::kj::_::hasSubstring(e->getDescription(), message), \ + "exception description didn't contain expected substring", *e, ##__VA_ARGS__); \ + } else { \ diff --git a/misc/capnproto.nix b/misc/capnproto.nix index 4d09a2892..27bd8920a 100644 --- a/misc/capnproto.nix +++ b/misc/capnproto.nix @@ -35,6 +35,11 @@ stdenv.mkDerivation rec { sha256 = "sha256-LVdkqVBTeh8JZ1McdVNtRcnFVwEJRNjt0JV2l7RkuO8="; }; + patches = [ + # backport of https://github.com/capnproto/capnproto/pull/1810 + ./capnproto-promise-nodiscard.patch + ]; + nativeBuildInputs = [ cmake ]; propagatedBuildInputs = [ openssl