From a0c663e7b40c1e6fe4c22ad679a21857981ea7f8 Mon Sep 17 00:00:00 2001 From: Rye Date: Wed, 28 Oct 2026 01:13:44 -0400 Subject: [PATCH 1/2] A thread pool can be let go of while its threads run A pool is closed by joining its threads, and destroying one closes it, so a pool whose thread is stuck -- in a write to a disk that does not answer -- holds whoever closes it for as long as the disk does. The settings files' writer is to run on a pool of one, and the plan for it has a close that waits two seconds and then leaves a stuck writer to finish on its own; nothing in a pool allowed that. ThreadPoolBase::abandon() takes the pool from its owner, in place of destroying it. It closes the queue and stops listening for the viewer's shutdown, there on the owner's thread; then, if every thread has finished, it joins them and destroys the pool, and otherwise lets them go, and the last of them to finish destroys the pool. close() joins only while the queue is open, so abandon() joins finished threads itself. The count of a pool's running threads, and the pool abandon() left to them, are kept in a state the threads share with the pool and keep once it is gone, because a pool's own run() may destroy it: the Win32 window's thread deletes its pool as it ends, once wakeAndDestroy() has let it go. A thread touches nothing of the pool after run() returns but what abandon() left it to destroy. A thread is counted before it starts, and not counted if it fails to. A pool destroyed on its own thread must not reach the event pumps, which are the main thread's: the destructor now stops listening only if the pool listened, which only one that shuts down with the viewer does. threadpool_test gains six cases beside the scheduler's: a pool of three stuck threads let go at once and destroyed by the last of them once released; one never started, destroyed at once; one whose queue's work still runs after it is let go; one listening for the shutdown that stops; a pool its own run() destroys, kept rather than freed, whose bytes nothing writes after; and a thread woken by a ping in under half a millisecond at the median, which the scheduler's own cases, of idle workers and timed waits, leave out. The wake is timed only where no sanitizer slows the run: under AddressSanitizer, beside the other tests on a CI runner, one took 3 ms. The first three fail with the last thread's deletion taken out, the fourth with the listener left; the fifth crashed before the shared state, and fails on any write to the pool after run(); the sixth took 15.5 ms under the scheduler that slept a tick at a time. Co-Authored-By: Claude Opus 5.5 --- indra/llcommon/tests/threadpool_test.cpp | 289 ++++++++++++++++++++++- indra/llcommon/threadpool.cpp | 111 ++++++++- indra/llcommon/threadpool.h | 22 ++ indra/test/altestsanitizer.h | 16 ++ 4 files changed, 426 insertions(+), 12 deletions(-) diff --git a/indra/llcommon/tests/threadpool_test.cpp b/indra/llcommon/tests/threadpool_test.cpp index 5d4653b434..4ffc4f8eb5 100644 --- a/indra/llcommon/tests/threadpool_test.cpp +++ b/indra/llcommon/tests/threadpool_test.cpp @@ -1,6 +1,8 @@ /** * @file threadpool_test.cpp - * @brief A pool's worker sleeps until work or a deadline wakes it. + * @brief A pool's worker sleeps until work or a deadline wakes it, and a pool + * let go of while its threads still run is destroyed by the last of + * them. * * $LicenseInfo:firstyear=2026&license=viewerlgpl$ * Alchemy Viewer Source Code @@ -26,10 +28,20 @@ #include "../threadpool.h" +#include "../llcond.h" +#include "../llcoros.h" +#include "../llevents.h" +#include "../llthread.h" +#include "../lltimer.h" + #if LL_WINDOWS #include "../llwin32headers.h" #endif +#include "altestsanitizer.h" + +#include LLCOROS_MUTEX_HEADER +#include LLCOROS_CONDVAR_HEADER #include #include @@ -38,6 +50,7 @@ #include #include #include +#include #include #include @@ -138,4 +151,278 @@ namespace CHECK(median < 0.0015); } #endif + + // A pool whose thread answers each ping, waiting between them as a + // worker waits for work: on a condition of the coroutines' own. + class EchoPool final : public LL::ThreadPool + { + public: + explicit EchoPool(const std::string& name) + : LL::ThreadPool(name, 1, 1, /*auto_shutdown*/ false, /*fixed_width*/ true) + { + } + + ~EchoPool() override + { + { + LLCoros::LockType lock(mMutex); + mStop = true; + } + mWake.notify_all(); + close(); + } + + // Asks the thread for an answer and spins until it comes: seconds. + F64 ping() + { + U32 asked = 0; + { + LLCoros::LockType lock(mMutex); + asked = ++mAsked; + } + LLTimer timer; + mWake.notify_one(); + while (mAnswered.load() != asked && timer.getElapsedTimeF64() < 5.0) + { + LLThread::yield(); + } + return timer.getElapsedTimeF64(); + } + + void run() override + { + LLCoros::LockType lock(mMutex); + while (!mStop) + { + mWake.wait(lock, [this] { return mStop || mAsked != mAnswered.load(); }); + mAnswered = mAsked; + } + } + + private: + LLCoros::Mutex mMutex; + LLCoros::ConditionVariable mWake; + U32 mAsked = 0; + std::atomic mAnswered{ 0 }; + bool mStop = false; + }; + + TEST_CASE("threadpool: a pool's thread wakes as soon as work is posted to it", "[threadpool]") + { + EchoPool pool("threadpool_test_echo"); + pool.start(); + std::vector waits; + for (int i = 0; i < 51; ++i) + { + waits.push_back(pool.ping()); + } + std::nth_element(waits.begin(), waits.begin() + waits.size() / 2, waits.end()); + const F64 median = waits[waits.size() / 2]; + INFO("median wake: " << median * 1000.0 << " ms"); + // A thread that looks for work only once a tick of the system timer + // answers half a tick late on average: 7.8 ms at Windows' default. A + // sanitizer's run is too slow and too loaded to time a wake by: one + // took 3 ms beside the other tests on a CI runner. + if constexpr (!ll_test::SLOW_SANITIZER) + { + CHECK(median < 0.0005); + } + } + + // Whether a pool of this name is alive. + bool live(const std::string& name) + { + for (const std::string& key : LL::ThreadPoolBase::key_snapshot()) + { + if (key == name) + { + return true; + } + } + return false; + } + + // Waits for a pool of this name to be gone, a while at most. + bool waitGone(const std::string& name) + { + LLTimer timer; + timer.setTimerExpirySec(10.f); + while (live(name)) + { + if (timer.hasExpired()) + { + return false; + } + ms_sleep(5); + } + return true; + } + + // A pool whose threads wait to be let go, as one stuck in a call to a + // disk that does not answer would. + class StuckPool final : public LL::ThreadPool + { + public: + StuckPool(const std::string& name, size_t width, LLOneShotCond& release, std::atomic& finished) + : LL::ThreadPool(name, width, 1, /*auto_shutdown*/ false, /*fixed_width*/ true), + mRelease(release), + mFinished(finished) + { + } + + ~StuckPool() override { close(); } + + void run() override + { + mRelease.wait(); + ++mFinished; + } + + private: + LLOneShotCond& mRelease; + std::atomic& mFinished; + }; + + TEST_CASE("threadpool: abandon returns at once and the last thread to finish destroys the pool", "[threadpool]") + { + const std::string name = "threadpool_test_stuck"; + LLOneShotCond release; + std::atomic finished{ 0 }; + { + auto pool = std::make_unique(name, 3, release, finished); + pool->start(); + LLTimer timer; + LL::ThreadPoolBase::abandon(std::move(pool)); + CHECK(timer.getElapsedTimeF32() < 1.f); + } + // The threads still run, and the pool is theirs. + CHECK(live(name)); + CHECK(finished == 0); + + release.set_all(); + CHECK(waitGone(name)); + CHECK(finished == 3); + } + + TEST_CASE("threadpool: abandon of a pool that never started destroys it at once", "[threadpool]") + { + const std::string name = "threadpool_test_unstarted"; + auto pool = std::make_unique(name, 2, 16, /*auto_shutdown*/ false, true); + REQUIRE(live(name)); + LL::ThreadPoolBase::abandon(std::move(pool)); + CHECK_FALSE(live(name)); + } + + TEST_CASE("threadpool: abandon closes the queue and its threads finish and take the pool", "[threadpool]") + { + const std::string name = "threadpool_test_queue"; + std::atomic ran{ 0 }; + { + auto pool = std::make_unique(name, 2, 16, /*auto_shutdown*/ false, true); + pool->start(); + for (int i = 0; i < 4; ++i) + { + REQUIRE(pool->getQueue().post([&ran] { ++ran; })); + } + LL::ThreadPoolBase::abandon(std::move(pool)); + } + CHECK(waitGone(name)); + // What was posted before the queue closed is still run: a closed + // queue is drained. + CHECK(ran == 4); + } + + TEST_CASE("threadpool: a pool listening for the viewer's shutdown stops on abandon", "[threadpool]") + { + const std::string name = "threadpool_test_listening"; + const std::string listener = "ThreadPool:" + name; + { + auto pool = std::make_unique(name, 1, 16, /*auto_shutdown*/ true, true); + pool->start(); + REQUIRE(LLEventPumps::instance().obtain("LLApp").getListener(listener).connected()); + LL::ThreadPoolBase::abandon(std::move(pool)); + } + CHECK_FALSE(LLEventPumps::instance().obtain("LLApp").getListener(listener).connected()); + CHECK(waitGone(name)); + } + + // Set as a thread's thread-local objects go, after everything its + // function did. + std::atomic sThreadGone{ false }; + + struct ThreadGone + { + ~ThreadGone() { sThreadGone = true; } + }; + + // Where a pool that destroyed itself was: kept rather than freed, with + // its bytes as its destructors left them, so that a write after shows. + void* sRemains = nullptr; + std::vector sAsDestroyed; + + // Lets SelfDeletingPool's thread end; outside the pool, which is gone + // once the thread wakes. + LLOneShotCond sLetGo; + + // A pool whose run() destroys it, as the Win32 window's thread does: its + // owner lets the thread go and closes the queue, and the thread deletes + // the pool as it ends. + class SelfDeletingPool final : public LL::ThreadPool + { + public: + explicit SelfDeletingPool(const std::string& name) + : LL::ThreadPool(name, 1, 1, /*auto_shutdown*/ false, /*fixed_width*/ true) + { + } + + static void* operator new(size_t size) { return ::operator new(size); } + + static void operator delete(void* block, size_t size) + { + sRemains = block; + const auto* bytes = static_cast(block); + sAsDestroyed.assign(bytes, bytes + size); + } + + // What LLWindowWin32Thread::wakeAndDestroy() does. + void letGo() + { + for (auto& pair : mThreads) + { + pair.second.detach(); + } + mQueue->close(); + sLetGo.set_all(); + } + + void run() override + { + thread_local ThreadGone gone; + (void)gone; + sLetGo.wait(); + delete this; + } + }; + + TEST_CASE("threadpool: a pool its own run() destroys is not touched after", "[threadpool]") + { + sThreadGone = false; + auto* pool = new SelfDeletingPool("threadpool_test_self_deleting"); + pool->start(); + // From here the thread may destroy the pool at any moment. + pool->letGo(); + + LLTimer timer; + timer.setTimerExpirySec(10.f); + while (!sThreadGone && !timer.hasExpired()) + { + ms_sleep(1); + } + REQUIRE(sThreadGone); + REQUIRE(sRemains); + const auto* now = static_cast(sRemains); + CHECK(std::equal(sAsDestroyed.begin(), sAsDestroyed.end(), now)); + ::operator delete(sRemains); + sRemains = nullptr; + } } diff --git a/indra/llcommon/threadpool.cpp b/indra/llcommon/threadpool.cpp index 546127e994..af71610975 100644 --- a/indra/llcommon/threadpool.cpp +++ b/indra/llcommon/threadpool.cpp @@ -18,11 +18,13 @@ // external library headers // other Linden headers #include "commoncontrol.h" +#include "llcoros.h" #include "llerror.h" #include "llevents.h" #include "llsd.h" #include "stringize.h" +#include LLCOROS_MUTEX_HEADER #include #include @@ -141,6 +143,15 @@ struct sleepy_robin: public boost::fibers::algo::round_robin /***************************************************************************** * ThreadPoolBase *****************************************************************************/ +struct LL::ThreadPoolBase::RunState +{ + LLCoros::Mutex mMutex; + // The threads started and not yet done. + size_t mRunning = 0; + // Set by abandon(): the pool, for the last of them to destroy. + ThreadPoolBase* mOrphan = nullptr; +}; + LL::ThreadPoolBase::ThreadPoolBase(const std::string& name, size_t threads, WorkQueueBase* queue, @@ -150,7 +161,8 @@ LL::ThreadPoolBase::ThreadPoolBase(const std::string& name, mName("ThreadPool:" + name), mThreadCount(fixed_width ? threads : getConfiguredWidth(name, threads)), mQueue(queue), - mAutomaticShutdown(auto_shutdown) + mAutomaticShutdown(auto_shutdown), + mRunState(std::make_shared()) {} void LL::ThreadPoolBase::start() @@ -172,15 +184,32 @@ void LL::ThreadPoolBase::start() std::string os_name{ mThreadCount > 1 ? stringize(':', (i+1)) : std::string() }; os_name.insert(0, getKey().substr(0, OS_NAME_MAX - os_name.size())); #endif - mThreads.emplace_back(tname, [this, tname, os_name]() - { - // Tracy names the OS thread too, cut to the same fifteen - // bytes, so it goes first and the short name is the one left - LL_PROFILER_SET_THREAD_NAME(tname.c_str()); - set_thread_name(os_name.c_str()); - set_thread_fp_mode(); - run(tname); - }); + // Counted before it starts, so that it cannot be done before it is + // counted, and not counted if it never starts. + { + LLCoros::LockType lock(mRunState->mMutex); + ++mRunState->mRunning; + } + try + { + mThreads.emplace_back(tname, [this, tname, os_name, state = mRunState]() + { + // Tracy names the OS thread too, cut to the same fifteen + // bytes, so it goes first and the short name is the one left + LL_PROFILER_SET_THREAD_NAME(tname.c_str()); + set_thread_name(os_name.c_str()); + set_thread_fp_mode(); + run(tname); + // The pool may be gone by now. + threadDone(state); + }); + } + catch (...) + { + LLCoros::LockType lock(mRunState->mMutex); + --mRunState->mRunning; + throw; + } } if (!mAutomaticShutdown) @@ -192,6 +221,7 @@ void LL::ThreadPoolBase::start() // Listen on "LLApp", and when the app is shutting down, close the queue // and join the workers. + mListening = true; LLEventPumps::instance().obtain("LLApp").listen( mName, [this](const LLSD& stat) @@ -210,10 +240,69 @@ void LL::ThreadPoolBase::start() LL::ThreadPoolBase::~ThreadPoolBase() { close(); - if (!LLEventPumps::wasDeleted()) + stopListening(); +} + +void LL::ThreadPoolBase::stopListening() +{ + if (mListening && !LLEventPumps::wasDeleted()) { LLEventPumps::instance().obtain("LLApp").stopListening(mName); } + mListening = false; +} + +void LL::ThreadPoolBase::abandon(std::unique_ptr pool) +{ + if (!pool) + { + return; + } + pool->mQueue->close(); + // Here, not on the pool's last thread: the event pumps are the main + // thread's. + pool->stopListening(); + { + // The state outlives the pool, which its last thread may destroy as + // soon as this lets go of the lock. + const std::shared_ptr state = pool->mRunState; + LLCoros::LockType lock(state->mMutex); + if (state->mRunning > 0) + { + LL_DEBUGS("ThreadPool") << pool->mName << " abandoned to its threads" << LL_ENDL; + for (auto& pair: pool->mThreads) + { + pair.second.detach(); + } + state->mOrphan = pool.release(); + return; + } + } + // Every thread has finished. close() joins only while the queue is + // open, so they are joined here, and the pool destroyed as it would + // have been. + for (auto& pair: pool->mThreads) + { + if (pair.second.joinable()) + { + pair.second.join(); + } + } +} + +//static +void LL::ThreadPoolBase::threadDone(const std::shared_ptr& state) +{ + ThreadPoolBase* orphan = nullptr; + { + LLCoros::LockType lock(state->mMutex); + if (--state->mRunning == 0) + { + orphan = state->mOrphan; + } + } + // An abandoned pool's threads were let go, so none is joined. + delete orphan; } void LL::ThreadPoolBase::close() diff --git a/indra/llcommon/threadpool.h b/indra/llcommon/threadpool.h index 1b3fb96c0d..7da5eb544e 100644 --- a/indra/llcommon/threadpool.h +++ b/indra/llcommon/threadpool.h @@ -59,6 +59,18 @@ namespace LL */ void close(); + /** + * abandon() lets a pool go whose threads may not finish soon -- one + * stuck in a call that does not return, on a disk that does not + * answer -- where close() and destroying it would wait for them. It + * closes the queue and stops listening for the viewer's shutdown; + * then, if every thread has finished, it destroys the pool at once, + * and otherwise the pool is its threads', and the last of them to + * finish destroys it, touching nothing of it after. Call it in place + * of destroying the pool, on the thread that would have. + */ + static void abandon(std::unique_ptr pool); + std::string getName() const { return mName; } size_t getWidth() const { return mThreads.size(); } @@ -95,10 +107,20 @@ namespace LL bool mAutomaticShutdown; private: + // What a pool's threads share with it, and keep once it is gone: a + // pool's own run() may destroy it, as the Win32 window's does. + struct RunState; + void run(const std::string& name); + // Each thread's last call, which touches nothing of the pool but + // what abandon() left it to destroy. + static void threadDone(const std::shared_ptr& state); + void stopListening(); std::string mName; size_t mThreadCount; + std::shared_ptr mRunState; + bool mListening = false; }; /** diff --git a/indra/test/altestsanitizer.h b/indra/test/altestsanitizer.h index 41323a347a..13ccb31cd4 100644 --- a/indra/test/altestsanitizer.h +++ b/indra/test/altestsanitizer.h @@ -65,6 +65,22 @@ namespace ll_test #else inline constexpr bool SANITIZER_ALLOCATOR = false; #endif + + // Whether every memory access goes through a sanitizer's checks, as + // AddressSanitizer's and ThreadSanitizer's do: a run several times slower, + // which CI runs beside other such tests, so a case that times something + // against the wall clock passes its bound over under it. +#if defined(__SANITIZE_ADDRESS__) || defined(__SANITIZE_THREAD__) + inline constexpr bool SLOW_SANITIZER = true; +#elif defined(__has_feature) +#if __has_feature(address_sanitizer) || __has_feature(thread_sanitizer) + inline constexpr bool SLOW_SANITIZER = true; +#else + inline constexpr bool SLOW_SANITIZER = false; +#endif +#else + inline constexpr bool SLOW_SANITIZER = false; +#endif } #endif // AL_ALTESTSANITIZER_H From 476c0e5e8999aa72ee10872eb52d4b19d12b58e9 Mon Sep 17 00:00:00 2001 From: Rye Date: Mon, 26 Oct 2026 21:18:32 -0400 Subject: [PATCH 2/2] Add aldb, the SQLite files settings and later the caches keep The viewer's settings are to move from XML written whole at exit onto a SQLite file, and the cache's index and the chat logs onto files of their own after them. aldb is what they share. Nothing links it yet. ALDatabase is one file. Its writes are jobs posted to a thread of the file's own, which writes whatever has queued in one BEGIN IMMEDIATE transaction, each job in a savepoint, so that a job that fails, or throws, is dropped alone and the rest commit. A failure that may pass (busy, locked, the disk full, an I/O error) stops the batch and tries it again, waiting from 100 ms up to 5 s; it is said once, through the log and the owner's onFailure, until a write next succeeds. Where such a failure makes SQLite undo the whole transaction, not the job alone, the jobs that ran before it are tried again with it, not dropped, and none runs on outside a transaction. A job may be posted with a completion, told on the writer thread whether it committed. flush() waits for what was posted for closeWait at most, and says whether it was written. Close writes what is queued and checkpoints the WAL into the file, and gives up after closeWait rather than hang an exit on a disk that does not answer. Reads take a connection from a pool of read-only ones. A file is kept by what losing its last writes would cost: Durable syncs every commit (and fullfsyncs on macOS), Journal survives the process dying, Regenerable is not synced and is mapped into memory. Each is a WAL file, or a DELETE-journal one where the file system has no shared memory to give the WAL its index. Its tables are a list of migrations by user_version, run in one transaction, a VACUUM INTO copy of the file kept before each and the last two copies kept; the version is read again under the transaction's lock, so that two viewers opening a new file at once make its tables once, and a file newer than the viewer is opened and left alone. Its application_id names what kind of file it is, and a file that carries another is refused, not written. A file this process may only read -- marked read-only, owned by another user, on read-only media -- is opened for reading, and the open says so: its reads run, and its writes are refused, said once. After a crash the owner asks for the file to be checked. The check waits for a file another viewer is writing, as every read does, and takes for damage only what says it is damaged, not a check that could not run. A damaged file is moved aside with its WAL, under a .corrupt-