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; } 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/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 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',