535 lines · cpp
1//========- unittests/Support/ThreadPools.cpp - ThreadPools.h tests --========//2//3// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.4// See https://llvm.org/LICENSE.txt for license information.5// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception6//7//===----------------------------------------------------------------------===//8 9#include "llvm/Support/ThreadPool.h"10#include "llvm/ADT/STLExtras.h"11#include "llvm/ADT/ScopeExit.h"12#include "llvm/ADT/SetVector.h"13#include "llvm/ADT/SmallVector.h"14#include "llvm/Config/llvm-config.h" // for LLVM_ENABLE_THREADS15#include "llvm/Support/CommandLine.h"16#include "llvm/Support/Program.h"17#include "llvm/Support/TargetSelect.h"18#include "llvm/Support/Threading.h"19#include "llvm/TargetParser/Host.h"20#include "llvm/TargetParser/Triple.h"21 22#ifdef _WIN3223#include "llvm/Support/Windows/WindowsSupport.h"24#endif25 26#include <chrono>27#include <thread>28 29#include "gtest/gtest.h"30 31namespace testing {32namespace internal {33// Specialize gtest construct to provide friendlier name in the output.34#if LLVM_ENABLE_THREADS35template <> std::string GetTypeName<llvm::StdThreadPool>() {36 return "llvm::StdThreadPool";37}38#endif39template <> std::string GetTypeName<llvm::SingleThreadExecutor>() {40 return "llvm::SingleThreadExecutor";41}42} // namespace internal43} // namespace testing44 45using namespace llvm;46 47// Fixture for the unittests, allowing to *temporarily* disable the unittests48// on a particular platform49template <typename ThreadPoolImpl> class ThreadPoolTest : public testing::Test {50 Triple Host;51 SmallVector<Triple::ArchType, 4> UnsupportedArchs;52 SmallVector<Triple::OSType, 4> UnsupportedOSs;53 SmallVector<Triple::EnvironmentType, 1> UnsupportedEnvironments;54 55protected:56 // This is intended for platform as a temporary "XFAIL"57 bool isUnsupportedOSOrEnvironment() {58 Triple Host(Triple::normalize(sys::getProcessTriple()));59 60 if (find(UnsupportedEnvironments, Host.getEnvironment()) !=61 UnsupportedEnvironments.end())62 return true;63 64 if (is_contained(UnsupportedOSs, Host.getOS()))65 return true;66 67 if (is_contained(UnsupportedArchs, Host.getArch()))68 return true;69 70 return false;71 }72 73 ThreadPoolTest() {74 // Add unsupported configuration here, example:75 // UnsupportedArchs.push_back(Triple::x86_64);76 77 // See https://llvm.org/bugs/show_bug.cgi?id=2582978 UnsupportedArchs.push_back(Triple::ppc64le);79 UnsupportedArchs.push_back(Triple::ppc64);80 }81 82 /// Make sure this thread not progress faster than the main thread.83 void waitForMainThread() { waitForPhase(1); }84 85 /// Set the readiness of the main thread.86 void setMainThreadReady() { setPhase(1); }87 88 /// Wait until given phase is set using setPhase(); first "main" phase is 1.89 /// See also PhaseResetHelper below.90 void waitForPhase(int Phase) {91 std::unique_lock<std::mutex> LockGuard(CurrentPhaseMutex);92 CurrentPhaseCondition.wait(93 LockGuard, [&] { return CurrentPhase == Phase || CurrentPhase < 0; });94 }95 /// If a thread waits on another phase, the test could bail out on a failed96 /// assertion and ThreadPool destructor would wait() on all threads, which97 /// would deadlock on the task waiting. Create this helper to automatically98 /// reset the phase and unblock such threads.99 struct PhaseResetHelper {100 PhaseResetHelper(ThreadPoolTest *test) : test(test) {}101 ~PhaseResetHelper() { test->setPhase(-1); }102 ThreadPoolTest *test;103 };104 105 /// Advance to the given phase.106 void setPhase(int Phase) {107 {108 std::unique_lock<std::mutex> LockGuard(CurrentPhaseMutex);109 assert(Phase == CurrentPhase + 1 || Phase < 0);110 CurrentPhase = Phase;111 }112 CurrentPhaseCondition.notify_all();113 }114 115 void SetUp() override { CurrentPhase = 0; }116 117 SmallVector<llvm::BitVector, 0> RunOnAllSockets(ThreadPoolStrategy S);118 119 std::condition_variable CurrentPhaseCondition;120 std::mutex CurrentPhaseMutex;121 int CurrentPhase; // -1 = error, 0 = setup, 1 = ready, 2+ = custom122};123 124using ThreadPoolImpls = ::testing::Types<125#if LLVM_ENABLE_THREADS126 StdThreadPool,127#endif128 SingleThreadExecutor>;129 130TYPED_TEST_SUITE(ThreadPoolTest, ThreadPoolImpls, );131 132#define CHECK_UNSUPPORTED() \133 do { \134 if (this->isUnsupportedOSOrEnvironment()) \135 GTEST_SKIP(); \136 } while (0);137 138TYPED_TEST(ThreadPoolTest, AsyncBarrier) {139 CHECK_UNSUPPORTED();140 // test that async & barrier work together properly.141 142 std::atomic_int checked_in{0};143 144 TypeParam Pool;145 for (size_t i = 0; i < 5; ++i) {146 Pool.async([this, &checked_in] {147 this->waitForMainThread();148 ++checked_in;149 });150 }151 ASSERT_EQ(0, checked_in);152 this->setMainThreadReady();153 Pool.wait();154 ASSERT_EQ(5, checked_in);155}156 157static void TestFunc(std::atomic_int &checked_in, int i) { checked_in += i; }158 159TYPED_TEST(ThreadPoolTest, AsyncBarrierArgs) {160 CHECK_UNSUPPORTED();161 // Test that async works with a function requiring multiple parameters.162 std::atomic_int checked_in{0};163 164 DefaultThreadPool Pool;165 for (size_t i = 0; i < 5; ++i) {166 Pool.async(TestFunc, std::ref(checked_in), i);167 }168 Pool.wait();169 ASSERT_EQ(10, checked_in);170}171 172TYPED_TEST(ThreadPoolTest, Async) {173 CHECK_UNSUPPORTED();174 DefaultThreadPool Pool;175 std::atomic_int i{0};176 Pool.async([this, &i] {177 this->waitForMainThread();178 ++i;179 });180 Pool.async([&i] { ++i; });181 ASSERT_NE(2, i.load());182 this->setMainThreadReady();183 Pool.wait();184 ASSERT_EQ(2, i.load());185}186 187TYPED_TEST(ThreadPoolTest, AsyncMoveOnly) {188 CHECK_UNSUPPORTED();189 DefaultThreadPool Pool;190 std::promise<int> p;191 std::future<int> f = p.get_future();192 Pool.async([this, p = std::move(p)]() mutable {193 this->waitForMainThread();194 p.set_value(42);195 });196 this->setMainThreadReady();197 Pool.wait();198 ASSERT_EQ(42, f.get());199}200 201TYPED_TEST(ThreadPoolTest, AsyncRAIICaptures) {202 CHECK_UNSUPPORTED();203 DefaultThreadPool Pool(hardware_concurrency(2));204 205 // We use a task group and a non-atomic value to stress test that the chaining206 // of tasks via a captured RAII object in fact chains and synchronizes within207 // a group.208 ThreadPoolTaskGroup Group(Pool);209 int value = 0;210 211 // Create an RAII object that when destroyed schedules more work. This makes212 // it easy to check that the RAII is resolved at the same point as a task runs213 // on the thread pool.214 auto schedule_next = llvm::make_scope_exit([&Group, &value] {215 // We sleep before scheduling the final task to make it much more likely216 // that an incorrect implementation actually exbitits a bug. Without the217 // sleep, we may get "lucky" and have the second task finish before the218 // assertion below fails even with an incorrect implementaiton. The219 // sleep is making _failures_ more reliable, it is not needed for220 // correctness and this test should only flakily _pass_, never flakily221 // fail.222 std::this_thread::sleep_for(std::chrono::milliseconds(10));223 Group.async([&value] { value = 42; });224 });225 226 // Now schedule the initial task, moving the RAII object to schedule the final227 // task into its captures.228 Group.async([schedule_next = std::move(schedule_next)]() {229 // Nothing to do here, the captured RAII object does the work.230 });231 232 // Both tasks should complete here, synchronizing with the read of value.233 Group.wait();234 ASSERT_EQ(42, value);235}236 237TYPED_TEST(ThreadPoolTest, GetFuture) {238 CHECK_UNSUPPORTED();239 DefaultThreadPool Pool(hardware_concurrency(2));240 std::atomic_int i{0};241 Pool.async([this, &i] {242 this->waitForMainThread();243 ++i;244 });245 // Force the future using get()246 Pool.async([&i] { ++i; }).get();247 ASSERT_NE(2, i.load());248 this->setMainThreadReady();249 Pool.wait();250 ASSERT_EQ(2, i.load());251}252 253TYPED_TEST(ThreadPoolTest, GetFutureWithResult) {254 CHECK_UNSUPPORTED();255 DefaultThreadPool Pool(hardware_concurrency(2));256 auto F1 = Pool.async([] { return 1; });257 auto F2 = Pool.async([] { return 2; });258 259 this->setMainThreadReady();260 Pool.wait();261 ASSERT_EQ(1, F1.get());262 ASSERT_EQ(2, F2.get());263}264 265TYPED_TEST(ThreadPoolTest, GetFutureWithResultAndArgs) {266 CHECK_UNSUPPORTED();267 DefaultThreadPool Pool(hardware_concurrency(2));268 auto Fn = [](int x) { return x; };269 auto F1 = Pool.async(Fn, 1);270 auto F2 = Pool.async(Fn, 2);271 272 this->setMainThreadReady();273 Pool.wait();274 ASSERT_EQ(1, F1.get());275 ASSERT_EQ(2, F2.get());276}277 278TYPED_TEST(ThreadPoolTest, PoolDestruction) {279 CHECK_UNSUPPORTED();280 // Test that we are waiting on destruction281 std::atomic_int checked_in{0};282 {283 DefaultThreadPool Pool;284 for (size_t i = 0; i < 5; ++i) {285 Pool.async([this, &checked_in] {286 this->waitForMainThread();287 ++checked_in;288 });289 }290 ASSERT_EQ(0, checked_in);291 this->setMainThreadReady();292 }293 ASSERT_EQ(5, checked_in);294}295 296// Check running tasks in different groups.297TYPED_TEST(ThreadPoolTest, Groups) {298 CHECK_UNSUPPORTED();299 // Need at least two threads, as the task in group2300 // might block a thread until all tasks in group1 finish.301 ThreadPoolStrategy S = hardware_concurrency(2);302 if (S.compute_thread_count() < 2)303 GTEST_SKIP();304 DefaultThreadPool Pool(S);305 typename TestFixture::PhaseResetHelper Helper(this);306 ThreadPoolTaskGroup Group1(Pool);307 ThreadPoolTaskGroup Group2(Pool);308 309 // Check that waiting for an empty group is a no-op.310 Group1.wait();311 312 std::atomic_int checked_in1{0};313 std::atomic_int checked_in2{0};314 315 for (size_t i = 0; i < 5; ++i) {316 Group1.async([this, &checked_in1] {317 this->waitForMainThread();318 ++checked_in1;319 });320 }321 Group2.async([this, &checked_in2] {322 this->waitForPhase(2);323 ++checked_in2;324 });325 ASSERT_EQ(0, checked_in1);326 ASSERT_EQ(0, checked_in2);327 // Start first group and wait for it.328 this->setMainThreadReady();329 Group1.wait();330 ASSERT_EQ(5, checked_in1);331 // Second group has not yet finished, start it and wait for it.332 ASSERT_EQ(0, checked_in2);333 this->setPhase(2);334 Group2.wait();335 ASSERT_EQ(5, checked_in1);336 ASSERT_EQ(1, checked_in2);337}338 339// Check recursive tasks.340TYPED_TEST(ThreadPoolTest, RecursiveGroups) {341 CHECK_UNSUPPORTED();342 DefaultThreadPool Pool;343 ThreadPoolTaskGroup Group(Pool);344 345 std::atomic_int checked_in1{0};346 347 for (size_t i = 0; i < 5; ++i) {348 Group.async([this, &Pool, &checked_in1] {349 this->waitForMainThread();350 351 ThreadPoolTaskGroup LocalGroup(Pool);352 353 // Check that waiting for an empty group is a no-op.354 LocalGroup.wait();355 356 std::atomic_int checked_in2{0};357 for (size_t i = 0; i < 5; ++i) {358 LocalGroup.async([&checked_in2] { ++checked_in2; });359 }360 LocalGroup.wait();361 ASSERT_EQ(5, checked_in2);362 363 ++checked_in1;364 });365 }366 ASSERT_EQ(0, checked_in1);367 this->setMainThreadReady();368 Group.wait();369 ASSERT_EQ(5, checked_in1);370}371 372TYPED_TEST(ThreadPoolTest, RecursiveWaitDeadlock) {373 CHECK_UNSUPPORTED();374 ThreadPoolStrategy S = hardware_concurrency(2);375 if (S.compute_thread_count() < 2)376 GTEST_SKIP();377 DefaultThreadPool Pool(S);378 typename TestFixture::PhaseResetHelper Helper(this);379 ThreadPoolTaskGroup Group(Pool);380 381 // Test that a thread calling wait() for a group and is waiting for more tasks382 // returns when the last task finishes in a different thread while the waiting383 // thread was waiting for more tasks to process while waiting.384 385 // Task A runs in the first thread. It finishes and leaves386 // the background thread waiting for more tasks.387 Group.async([this] {388 this->waitForMainThread();389 this->setPhase(2);390 });391 // Task B is run in a second thread, it launches yet another392 // task C in a different group, which will be handled by the waiting393 // thread started above.394 Group.async([this, &Pool] {395 this->waitForPhase(2);396 ThreadPoolTaskGroup LocalGroup(Pool);397 LocalGroup.async([this] {398 this->waitForPhase(3);399 // Give the other thread enough time to check that there's no task400 // to process and suspend waiting for a notification. This is indeed racy,401 // but probably the best that can be done.402 std::this_thread::sleep_for(std::chrono::milliseconds(10));403 });404 // And task B only now will wait for the tasks in the group (=task C)405 // to finish. This test checks that it does not deadlock. If the406 // `NotifyGroup` handling in ThreadPool::processTasks() didn't take place,407 // this task B would be stuck waiting for tasks to arrive.408 this->setPhase(3);409 LocalGroup.wait();410 });411 this->setMainThreadReady();412 Group.wait();413}414 415#if LLVM_ENABLE_THREADS == 1416 417// FIXME: Skip some tests below on non-Windows because multi-socket systems418// were not fully tested on Unix yet, and llvm::get_thread_affinity_mask()419// isn't implemented for Unix (need AffinityMask in Support/Unix/Program.inc).420#ifdef _WIN32421 422template <typename ThreadPoolImpl>423SmallVector<llvm::BitVector, 0>424ThreadPoolTest<ThreadPoolImpl>::RunOnAllSockets(ThreadPoolStrategy S) {425 llvm::SetVector<llvm::BitVector> ThreadsUsed;426 std::mutex Lock;427 {428 std::condition_variable AllThreads;429 std::mutex AllThreadsLock;430 unsigned Active = 0;431 432 DefaultThreadPool Pool(S);433 for (size_t I = 0; I < S.compute_thread_count(); ++I) {434 Pool.async([&] {435 {436 std::lock_guard<std::mutex> Guard(AllThreadsLock);437 ++Active;438 AllThreads.notify_one();439 }440 this->waitForMainThread();441 std::lock_guard<std::mutex> Guard(Lock);442 auto Mask = llvm::get_thread_affinity_mask();443 ThreadsUsed.insert(Mask);444 });445 }446 EXPECT_EQ(true, ThreadsUsed.empty());447 {448 std::unique_lock<std::mutex> Guard(AllThreadsLock);449 AllThreads.wait(Guard,450 [&]() { return Active == S.compute_thread_count(); });451 }452 this->setMainThreadReady();453 }454 return ThreadsUsed.takeVector();455}456 457TYPED_TEST(ThreadPoolTest, AllThreads_UseAllRessources) {458 CHECK_UNSUPPORTED();459 // After Windows 11, the OS is free to deploy the threads on any CPU socket.460 // We cannot relibly ensure that all thread affinity mask are covered,461 // therefore this test should not run.462 if (llvm::RunningWindows11OrGreater())463 GTEST_SKIP();464 auto ThreadsUsed = this->RunOnAllSockets({});465 ASSERT_EQ(llvm::get_cpus(), ThreadsUsed.size());466}467 468TYPED_TEST(ThreadPoolTest, AllThreads_OneThreadPerCore) {469 CHECK_UNSUPPORTED();470 // After Windows 11, the OS is free to deploy the threads on any CPU socket.471 // We cannot relibly ensure that all thread affinity mask are covered,472 // therefore this test should not run.473 if (llvm::RunningWindows11OrGreater())474 GTEST_SKIP();475 auto ThreadsUsed =476 this->RunOnAllSockets(llvm::heavyweight_hardware_concurrency());477 ASSERT_EQ(llvm::get_cpus(), ThreadsUsed.size());478}479 480// From TestMain.cpp.481extern const char *TestMainArgv0;482 483// Just a reachable symbol to ease resolving of the executable's path.484static cl::opt<std::string> ThreadPoolTestStringArg1("thread-pool-string-arg1");485 486#ifdef _WIN32487#define setenv(name, var, ignore) _putenv_s(name, var)488#endif489 490TYPED_TEST(ThreadPoolTest, AffinityMask) {491 CHECK_UNSUPPORTED();492 493 // Skip this test if less than 4 threads are available.494 if (llvm::hardware_concurrency().compute_thread_count() < 4)495 GTEST_SKIP();496 497 using namespace llvm::sys;498 if (getenv("LLVM_THREADPOOL_AFFINITYMASK")) {499 auto ThreadsUsed = this->RunOnAllSockets({});500 // Ensure the threads only ran on CPUs 0-3.501 // NOTE: Don't use ASSERT* here because this runs in a subprocess,502 // and will show up as un-executed in the parent.503 assert(llvm::all_of(ThreadsUsed,504 [](auto &T) { return T.getData().front() < 16UL; }) &&505 "Threads ran on more CPUs than expected! The affinity mask does not "506 "seem to work.");507 return;508 }509 std::string Executable =510 sys::fs::getMainExecutable(TestMainArgv0, &ThreadPoolTestStringArg1);511 const auto *TestInfo = testing::UnitTest::GetInstance()->current_test_info();512 std::string Arg = (Twine("--gtest_filter=") + TestInfo->test_suite_name() +513 "." + TestInfo->name())514 .str();515 StringRef argv[] = {Executable, Arg};516 517 // Add environment variable to the environment of the child process.518 int Res = setenv("LLVM_THREADPOOL_AFFINITYMASK", "1", false);519 ASSERT_EQ(0, Res);520 521 std::string Error;522 bool ExecutionFailed;523 BitVector Affinity;524 Affinity.resize(4);525 Affinity.set(0, 4); // Use CPUs 0,1,2,3.526 int Ret = sys::ExecuteAndWait(Executable, argv, {}, {}, 0, 0, &Error,527 &ExecutionFailed, nullptr, &Affinity);528 Res = setenv("LLVM_THREADPOOL_AFFINITYMASK", "", false);529 ASSERT_EQ(0, Res);530 ASSERT_EQ(0, Ret);531}532 533#endif // #ifdef _WIN32534#endif // #if LLVM_ENABLE_THREADS == 1535