diff --git a/.gitmodules b/.gitmodules index 5bebb9c2..ecfa19d1 100644 --- a/.gitmodules +++ b/.gitmodules @@ -34,3 +34,6 @@ [submodule "3rdparty/asio"] path = 3rdparty/asio url = https://github.com/chriskohlhoff/asio.git +[submodule "3rdparty/libfork"] + path = 3rdparty/libfork + url = https://github.com/ConorWilliams/libfork.git diff --git a/3rdparty/libfork b/3rdparty/libfork new file mode 160000 index 00000000..9b2b844a --- /dev/null +++ b/3rdparty/libfork @@ -0,0 +1 @@ +Subproject commit 9b2b844a5f4ed19296be643fac2f5fc04008eab8 diff --git a/CMakeLists.txt b/CMakeLists.txt index 3ff0e1a3..be4db22c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -373,6 +373,13 @@ set(MOBILEGL_INCLUDE_DIR # MG_Util/Async/ShaderCompilePool.cpp includes it, and it stays behind that file's # pimpl so no consumer target needs this path. ${CMAKE_SOURCE_DIR}/3rdparty/asio/asio/include + # The second shader-compile execution engine (MOBILEGL_ASYNC_POOL=libfork), on the + # same terms as Asio above: header-only, no add_subdirectory (its CMakeLists only + # declares an INTERFACE target plus install/test scaffolding we do not want), no link + # target, and reachable from exactly one translation unit. libfork's own + # target_compile_features asks for cxx_std_23, which this project already sets + # globally, so its C++20 coroutines need no per-source standard override. + ${CMAKE_SOURCE_DIR}/3rdparty/libfork/include ) add_library(${CMAKE_PROJECT_NAME} SHARED diff --git a/MobileGL/Config.h b/MobileGL/Config.h index 087ae783..6e846937 100644 --- a/MobileGL/Config.h +++ b/MobileGL/Config.h @@ -66,6 +66,12 @@ namespace MobileGL::MG_Config { // - DISPLAY: X11 session variable, not MobileGL configuration. // - MOBILEGL_LOG_FILE_PATH: log-file init runs before MG_ConfigLoader::Init // (see MG_Util/Debug/Log.cpp). + // - MOBILEGL_ASYNC_POOL: a ShaderCompilePool is constructed by binaries that never call + // MobileGL::Initialize() and so never run MG_ConfigLoader::Init - MG_Test's + // JobNodeTest builds pools directly, and it is the suite that runs the whole async + // matrix against both execution engines. Mirroring it here would resolve to the + // default in exactly the tests that exist to tell the engines apart (see + // MG_Util/Async/ShaderCompilePool.cpp, DetectAsyncPoolEngine). struct FeaturesTable { // MOBILEGL_DISABLE_TIMERQUERY: do not advertise or use GPU timer queries. Bool DisableTimerQuery = false; diff --git a/MobileGL/MG_Test/Util/AsyncPoolBench.cpp b/MobileGL/MG_Test/Util/AsyncPoolBench.cpp new file mode 100644 index 00000000..5ab6dce8 --- /dev/null +++ b/MobileGL/MG_Test/Util/AsyncPoolBench.cpp @@ -0,0 +1,368 @@ +// MobileGL - MobileGL/MG_Test/Util/AsyncPoolBench.cpp +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// A head-to-head harness for the two ShaderCompilePool execution engines +// (MOBILEGL_ASYNC_POOL=asio|libfork). Not a gtest: it measures one wall-clock interval per +// process, because most of what it drives is memoized per process (the shader preprocess +// cache and the compile-adoption map both live for the life of the GL context), so a second +// timed repetition inside one process would measure the cache, not the compiler. The driver +// script re-executes the binary for every repetition instead. +// +// Two modes: +// +// corpus - the REAL frontend path. glCreateShader/glShaderSource are done untimed, then +// the clock starts and glCompileShader/glLinkProgram submit every job, and stops +// once glGetProgramiv(GL_LINK_STATUS) has joined all of them. That is exactly the +// first-submit-to-all-joined interval a shaderpack load pays. +// +// micro - N trivial JobNodes straight through ShaderCompilePool::Post, isolating the +// executor's own dispatch overhead from any workload contention. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "Includes.h" +#include "Init.h" +#include + +#include +#include +#include + +using namespace MobileGL; +using namespace MobileGL::MG_Util::Async; +namespace GLImpl = MobileGL::MG_Impl::GLImpl; +namespace fs = std::filesystem; + +namespace { + using Clock = std::chrono::steady_clock; + + double MillisSince(const Clock::time_point start) { + return std::chrono::duration(Clock::now() - start).count(); + } + + GLenum StageFromExtension(const std::string& ext) { + if (ext == ".vert") return GL_VERTEX_SHADER; + if (ext == ".frag") return GL_FRAGMENT_SHADER; + if (ext == ".geom") return GL_GEOMETRY_SHADER; + if (ext == ".comp") return GL_COMPUTE_SHADER; + if (ext == ".tesc") return GL_TESS_CONTROL_SHADER; + if (ext == ".tese") return GL_TESS_EVALUATION_SHADER; + return 0; + } + + std::string ReadFile(const fs::path& path) { + std::ifstream in(path, std::ios::binary); + std::ostringstream buf; + buf << in.rdbuf(); + return buf.str(); + } + + struct CorpusShader { + std::string name; + std::string source; + GLenum stage = 0; + }; + + // One program's worth of the corpus: the trace's link group. Shaders are indices into + // the flat shader list, because a source shared by several programs must stay ONE entry + // - that sharing is what the compile-adoption map sees in the real path too. + struct CorpusProgram { + std::vector shaders; + }; + + struct Corpus { + std::vector shaders; + std::vector programs; + SizeT totalBytes = 0; + }; + + // Reads a corpus directory written by extract_corpus.py: one file per compiled shader, + // stage in the extension, plus manifest.txt naming the trace's link groups. + Corpus LoadCorpus(const fs::path& dir) { + Corpus corpus; + std::unordered_map byName; + + const auto intern = [&](const std::string& name) -> SizeT { + if (const auto it = byName.find(name); it != byName.end()) return it->second; + const fs::path path = dir / name; + if (!fs::exists(path)) return static_cast(-1); + CorpusShader shader; + shader.name = name; + shader.source = ReadFile(path); + shader.stage = StageFromExtension(path.extension().string()); + if (shader.stage == 0) return static_cast(-1); + corpus.totalBytes += shader.source.size(); + corpus.shaders.push_back(Move(shader)); + const SizeT index = corpus.shaders.size() - 1; + byName.emplace(name, index); + return index; + }; + + const fs::path manifest = dir / "manifest.txt"; + if (fs::exists(manifest)) { + std::ifstream in(manifest); + std::string line; + while (std::getline(in, line)) { + if (line.empty() || line[0] == '#') continue; + CorpusProgram program; + std::istringstream fields(line); + std::string name; + while (fields >> name) { + const SizeT index = intern(name); + if (index != static_cast(-1)) program.shaders.push_back(index); + } + if (!program.shaders.empty()) corpus.programs.push_back(Move(program)); + } + } + + // Anything in the directory the manifest never linked still gets compiled, as a + // program-less group, so the corpus on disk and the corpus measured are the same set. + std::vector leftovers; + for (const auto& entry : fs::directory_iterator(dir)) { + if (!entry.is_regular_file()) continue; + const std::string name = entry.path().filename().string(); + if (name == "manifest.txt") continue; + if (StageFromExtension(entry.path().extension().string()) == 0) continue; + if (byName.count(name) != 0) continue; + leftovers.push_back(entry.path()); + } + std::sort(leftovers.begin(), leftovers.end()); + for (const auto& path : leftovers) intern(path.filename().string()); + + return corpus; + } + + struct CorpusResult { + double submitMs = 0; // first glCompileShader -> last glLinkProgram returned + double joinMs = 0; // last submit -> every program joined + double totalMs = 0; // the number that matters: first submit -> all joined + SizeT linkFailures = 0; + SizeT compileFailures = 0; + }; + + CorpusResult RunCorpus(const Corpus& corpus) { + // ---- Untimed: create every GL object and stage every source ---------------------- + // glShaderSource is a memcpy into the shader object and glAttachShader is a pointer + // append; neither touches the pool. Keeping them outside the clock makes the measured + // interval exactly the compile+link critical path, which is what an application's + // loading screen waits on. + std::vector shaderNames(corpus.shaders.size(), 0); + for (SizeT i = 0; i < corpus.shaders.size(); ++i) { + const CorpusShader& shader = corpus.shaders[i]; + const GLuint name = GLImpl::CreateShader(shader.stage); + const GLchar* text = shader.source.c_str(); + const GLint length = static_cast(shader.source.size()); + GLImpl::ShaderSource(name, 1, &text, &length); + shaderNames[i] = name; + } + + std::vector programNames(corpus.programs.size(), 0); + for (SizeT p = 0; p < corpus.programs.size(); ++p) { + const GLuint program = GLImpl::CreateProgram(); + for (const SizeT shaderIndex : corpus.programs[p].shaders) { + GLImpl::AttachShader(program, shaderNames[shaderIndex]); + } + programNames[p] = program; + } + + // ---- Timed ------------------------------------------------------------------------ + const Clock::time_point start = Clock::now(); + + // Submission order follows the trace: a program's shaders, then its link. That order + // is what exercises ProgramLinkTask::SubmitAfter's dependency chaining rather than a + // flat burst of independent compiles. + std::vector submitted(corpus.shaders.size(), false); + for (SizeT p = 0; p < corpus.programs.size(); ++p) { + for (const SizeT shaderIndex : corpus.programs[p].shaders) { + if (submitted[shaderIndex]) continue; + submitted[shaderIndex] = true; + GLImpl::CompileShader(shaderNames[shaderIndex]); + } + GLImpl::LinkProgram(programNames[p]); + } + for (SizeT i = 0; i < corpus.shaders.size(); ++i) { + if (submitted[i]) continue; + submitted[i] = true; + GLImpl::CompileShader(shaderNames[i]); + } + + const Clock::time_point submitted_at = Clock::now(); + + CorpusResult result; + // GL_LINK_STATUS is a joining query (GL_COMPLETION_STATUS_KHR is the one that must + // not join), so this loop is the all-joined barrier. + for (const GLuint program : programNames) { + GLint status = 0; + GLImpl::GetProgramiv(program, GL_LINK_STATUS, &status); + if (status == GL_FALSE) ++result.linkFailures; + } + for (const GLuint shader : shaderNames) { + GLint status = 0; + GLImpl::GetShaderiv(shader, GL_COMPILE_STATUS, &status); + if (status == GL_FALSE) ++result.compileFailures; + } + + result.totalMs = MillisSince(start); + result.submitMs = std::chrono::duration(submitted_at - start).count(); + result.joinMs = result.totalMs - result.submitMs; + + for (const GLuint program : programNames) GLImpl::DeleteProgram(program); + for (const GLuint shader : shaderNames) GLImpl::DeleteShader(shader); + return result; + } + + // ---- Executor microbenchmark ---------------------------------------------------------- + // The body is deliberately near-empty: what is being measured is Post -> engine -> + // RunOnWorker -> next dispatch, i.e. the executor's own cost per job, with no compiler + // work to hide it. + // + // The barrier is an all-jobs-ran latch, and it has to be. This bench used to stop the + // clock at StopAndDrain(), which is not a "wait for everything" - it is the teardown path, + // and its contract is to ABANDON whatever the budget has not dispatched yet (see + // ShaderCompilePool::StopAndDrain, and the JobNodeTest case that pins exactly that). With + // 100k jobs behind a budget of N, most of them were therefore cancelled rather than run, + // and the fraction that survived was decided by how fast the engine drained the queue + // relative to the posting loop - i.e. by the very quantity under test. Measured on this + // machine at 8 workers: Asio ran 75,906 of 100,000 and libfork 99,998, and both were + // scored as if they had run 100,000. The reported "libfork is 1.36x faster" was libfork + // being charged for 32% more work than Asio. + class TrivialJob final : public JobNode { + public: + TrivialJob(std::atomic* sink, const Uint64 total, std::mutex* mutex, + std::condition_variable* cv) + : m_sink(sink), m_total(total), m_mutex(mutex), m_cv(cv) {} + + private: + void RunBody() override { + if (m_sink->fetch_add(1, std::memory_order_acq_rel) + 1 == m_total) { + // The last job wakes the timer. Under the lock, so the waiter cannot miss it + // between its predicate check and its wait. + const std::lock_guard lock(*m_mutex); + m_cv->notify_all(); + } + } + + std::atomic* m_sink; + Uint64 m_total; + std::mutex* m_mutex; + std::condition_variable* m_cv; + }; + + struct MicroResult { + double ms = 0; + Uint64 ran = 0; + }; + + MicroResult RunMicrobench(const Uint threads, const SizeT jobs) { + ShaderCompilePool pool(threads); + std::atomic counter{0}; + std::mutex mutex; + std::condition_variable cv; + const auto total = static_cast(jobs); + + // Nodes are allocated up front: MakeShared is not what is under test, and leaving it + // inside the loop would put an allocator on the critical path in front of the + // dispatch path this is meant to isolate. + std::vector> nodes; + nodes.reserve(jobs); + for (SizeT i = 0; i < jobs; ++i) { + nodes.push_back(MakeShared(&counter, total, &mutex, &cv)); + } + + const Clock::time_point start = Clock::now(); + for (auto& node : nodes) pool.Post(Move(node)); + { + std::unique_lock lock(mutex); + cv.wait(lock, [&] { return counter.load(std::memory_order_acquire) >= total; }); + } + const double ms = MillisSince(start); + + MicroResult result; + result.ms = ms; + result.ran = counter.load(std::memory_order_acquire); + return result; + } + + [[noreturn]] void Usage() { + std::fprintf(stderr, + "usage: AsyncPoolBench --corpus DIR\n" + " AsyncPoolBench --micro JOBS --threads N\n" + "env: MOBILEGL_ASYNC_POOL=asio|libfork, " + "MOBILEGL_ASYNC_SHADER_COMPILE_THREADS=N\n"); + std::exit(2); + } +} // namespace + +int main(int argc, char** argv) { + std::string corpusDir; + SizeT microJobs = 0; + Uint microThreads = 0; + + for (int i = 1; i < argc; ++i) { + const std::string arg = argv[i]; + const auto next = [&]() -> std::string { + if (i + 1 >= argc) Usage(); + return argv[++i]; + }; + if (arg == "--corpus") corpusDir = next(); + else if (arg == "--micro") microJobs = static_cast(std::stoull(next())); + else if (arg == "--threads") microThreads = static_cast(std::stoul(next())); + else Usage(); + } + if (corpusDir.empty() && microJobs == 0) Usage(); + + Initialize(); + + const AsyncPoolEngine engine = DetectAsyncPoolEngine(); + const char* engineName = AsyncPoolEngineName(engine); + + if (microJobs != 0) { + const Uint threads = microThreads != 0 ? microThreads : DetectShaderCompileThreadCount(); + const MicroResult result = RunMicrobench(threads, microJobs); + // `ran` is printed, not just checked, so that a run in which the arms did different + // amounts of work is visible in the results file rather than on a stderr the driver + // script redirects to /dev/null. ns_per_job divides by what actually ran. + std::printf("RESULT mode=micro engine=%s threads=%u jobs=%zu ran=%llu total_ms=%.3f " + "ns_per_job=%.1f\n", + engineName, threads, microJobs, static_cast(result.ran), + result.ms, result.ms * 1e6 / static_cast(result.ran)); + return result.ran == microJobs ? 0 : 1; + } + + const Corpus corpus = LoadCorpus(corpusDir); + if (corpus.shaders.empty()) { + std::fprintf(stderr, "AsyncPoolBench: no shaders found in %s\n", corpusDir.c_str()); + return 1; + } + + if (!AsyncShaderCompileActive()) { + std::fprintf(stderr, "AsyncPoolBench: asynchronous compilation is OFF; measuring the " + "inline path\n"); + } + + const CorpusResult result = RunCorpus(corpus); + const Uint threads = ShaderCompilePool::Get().GetThreadCount(); + + std::printf("RESULT mode=corpus engine=%s threads=%u corpus=%s shaders=%zu programs=%zu " + "bytes=%zu total_ms=%.3f submit_ms=%.3f join_ms=%.3f link_fail=%zu " + "compile_fail=%zu\n", + engineName, threads, corpusDir.c_str(), corpus.shaders.size(), + corpus.programs.size(), corpus.totalBytes, result.totalMs, result.submitMs, + result.joinMs, result.linkFailures, result.compileFailures); + return 0; +} diff --git a/MobileGL/MG_Test/Util/CMakeLists.txt b/MobileGL/MG_Test/Util/CMakeLists.txt index 2bc31b7d..08ddf3b5 100644 --- a/MobileGL/MG_Test/Util/CMakeLists.txt +++ b/MobileGL/MG_Test/Util/CMakeLists.txt @@ -10,11 +10,35 @@ target_include_directories(JobNodeTest PRIVATE ${MGL_ROOT}/MobileGL ) +# GTest::gtest, not GTest::gtest_main: JobNodeTest supplies its own main so that +# MOBILEGL_LOG_FILE_PATH is set before the first log write in the process. The engine +# -selection cases read the log back to assert that an unrecognized MOBILEGL_ASYNC_POOL value +# warns, and the desktop log sink is the file (MOBILEGL_LOG_ENABLE_CONSOLE is 0). target_link_libraries( JobNodeTest PRIVATE - GTest::gtest_main + GTest::gtest ${LINK_LIBRARIES} ) include(GoogleTest) gtest_discover_tests(JobNodeTest DISCOVERY_TIMEOUT 30 PROPERTIES LABELS unit) + +# The engine comparison harness. Deliberately NOT registered with add_test: it measures wall +# time, so it has no pass/fail verdict to give CI, and it is driven by a script that varies +# MOBILEGL_ASYNC_POOL and MOBILEGL_ASYNC_SHADER_COMPILE_THREADS across a matrix. It lives +# beside JobNodeTest because it drives the same pool through the same two engines; it links +# MobileGL_s for the real glCompileShader/glLinkProgram frontend path. +add_executable( + AsyncPoolBench + AsyncPoolBench.cpp +) + +target_include_directories(AsyncPoolBench PRIVATE + ${MGL_ROOT}/include + ${MGL_ROOT}/MobileGL +) + +target_link_libraries( + AsyncPoolBench PRIVATE + ${LINK_LIBRARIES} +) diff --git a/MobileGL/MG_Test/Util/JobNodeTest.cpp b/MobileGL/MG_Test/Util/JobNodeTest.cpp index 79fbc002..08ca5d06 100644 --- a/MobileGL/MG_Test/Util/JobNodeTest.cpp +++ b/MobileGL/MG_Test/Util/JobNodeTest.cpp @@ -9,8 +9,19 @@ #include #include +#include +#include +#include #include +#ifdef _WIN32 +#include +#define MGL_TEST_GETPID _getpid +#else +#include +#define MGL_TEST_GETPID getpid +#endif + #include "Includes.h" #include @@ -21,6 +32,29 @@ using namespace MobileGL; using namespace MobileGL::MG_Util::Async; namespace { + // Where this binary's MobileGL log lands, set by main() below. The engine-selection cases + // read it back: MobileGL's desktop log sink is the FILE, not the console + // (MOBILEGL_LOG_ENABLE_CONSOLE is 0 in Defines.h), so gtest's stdout capture would see + // nothing, and "unrecognized value warns" is a contract worth pinning rather than + // assuming - a silent fallback makes a misspelt engine name look exactly like an unset + // variable. + String g_logFilePath; + + // Log.cpp flushes the file after every line, so everything written before this call is + // already visible. + String ReadLogFrom(const std::streamoff offset) { + std::ifstream file(g_logFilePath, std::ios::binary); + if (!file) return {}; + file.seekg(offset); + return String((std::istreambuf_iterator(file)), std::istreambuf_iterator()); + } + + std::streamoff LogSize() { + std::error_code error; + const auto size = std::filesystem::file_size(g_logFilePath, error); + return error ? 0 : static_cast(size); + } + // Every test drives its own pool instance rather than ShaderCompilePool::Get(): the // process-wide pool is stopped permanently by StopAndDrain (that is the teardown // contract), so a test that drained the singleton would poison every test after it. @@ -73,6 +107,21 @@ namespace { Bool m_open = false; }; + // Live thread count of this process. Linux only - /proc/self/task has one entry per + // thread - and 0 where that is not available, which is how the one case that uses it + // decides to skip rather than to assert something it cannot see. + SizeT LiveThreadCount() { +#ifdef __linux__ + std::error_code error; + const auto count = static_cast( + std::distance(std::filesystem::directory_iterator("/proc/self/task", error), + std::filesystem::directory_iterator())); + return error ? 0 : count; +#else + return 0; +#endif + } + Bool WaitUntil(const std::function& predicate, const std::chrono::milliseconds timeout = std::chrono::seconds(10)) { const auto deadline = std::chrono::steady_clock::now() + timeout; @@ -89,11 +138,33 @@ namespace { // --------------------------------------------------------------------------------------- TEST(ShaderCompilePoolLifecycle, ConstructingAPoolStartsNoThreadUntilSomethingIsPosted) { + const SizeT before = LiveThreadCount(); + ShaderCompilePool pool(kTestThreads); EXPECT_EQ(pool.GetThreadCount(), kTestThreads); EXPECT_EQ(pool.GetMaxConcurrency(), kTestThreads); - // Nothing observable to assert about thread creation from here; what this pins is that - // construction is side-effect free and the pool destructs cleanly without ever running. + + if (before == 0) { + // No thread census on this platform. The rest still holds: construction is + // side-effect free and the pool destructs cleanly without ever having run. + SUCCEED(); + return; + } + + // "A build that never posts pays nothing" is a real requirement, not a stylistic one - + // asynchronous compilation can be switched off entirely, and a switched-off pool that + // still spawned its workers would cost every such process its threads and their stacks. + // Worth asserting rather than asserting-by-comment now that an engine's thread shape is + // selectable: the libfork engine starts its workers AND a dispatch thread of its own, so + // a regression here would cost more than it used to. + EXPECT_EQ(LiveThreadCount(), before) << "constructing a pool started " << (LiveThreadCount() - before) + << " thread(s) before anything was posted"; + + auto job = MakeShared(); + pool.Post(job); + job->Wait(); + EXPECT_GT(LiveThreadCount(), before) << "the first Post started no thread at all, so the engine did not " + "really run the job off the calling thread"; } TEST(ShaderCompilePoolLifecycle, StopAndDrainIsIdempotentAndSafeOnAnUnusedPool) { @@ -217,6 +288,94 @@ TEST(JobNodeSubmit, ManyJobsAllComplete) { } } +TEST(JobNodeSubmit, AJobBodyMayPostAnotherJobToTheSamePool) { + // The ProgramLinkTask::SubmitAfter shape, reduced to its scheduling core: the dependent is + // posted by whichever thread drove the dependency terminal, which for a job that finished + // on a worker is that WORKER. Every engine therefore has to accept a submission from + // inside its own pool. + // + // Not a hypothetical: libfork refuses this outright at its normal entry point + // (lf::schedule throws lf::schedule_in_worker, because a libfork worker may never block), + // which is why the libfork engine owns a dispatch thread of its own. Without this case a + // naive port passes every other test in the file and turns every dependency-released link + // job into a cancelled one on the real GL path. + ShaderCompilePool pool(kTestThreads); + + std::atomic innerSawPoolThread{false}; + auto inner = MakeShared( + [&](TestJob&) { innerSawPoolThread.store(ShaderCompilePool::IsPoolThread(), std::memory_order_release); }); + + std::atomic postedFromPoolThread{false}; + auto outer = MakeShared([&](TestJob&) { + postedFromPoolThread.store(ShaderCompilePool::IsPoolThread(), std::memory_order_release); + pool.Post(inner); + }); + + pool.Post(outer); + outer->Wait(); + inner->Wait(); + + EXPECT_TRUE(postedFromPoolThread.load()) << "the outer body did not run on a pool thread, so this case " + "did not exercise posting from inside the pool"; + EXPECT_TRUE(outer->IsComplete()); + // The load-bearing one: the inner job RAN. A dispatch the engine refused would have + // settled it Cancelled instead, and its body would never have executed. + EXPECT_TRUE(inner->IsComplete()) << "a job posted from a pool thread was not dispatched"; + EXPECT_FALSE(inner->IsCancelled()); + EXPECT_EQ(inner->ran.load(), 1u); + EXPECT_TRUE(innerSawPoolThread.load()); +} + +TEST(JobNodeSubmit, ABurstPostedFromInsideThePoolStillRunsInParallel) { + // The tail of a pack load: one compile job goes terminal and its continuations release + // several programs at once (ShaderCompileAdoptionMap lets one compile settle many), so a + // WORKER posts a burst into a pool that is otherwise idle. Every one of those posts clears + // the budget immediately, so the engine is handed `kBurst` runnable jobs from inside + // itself - and it has to spread them, not run them one behind another on the thread that + // submitted them. + // + // Asserting on peak concurrency rather than on wall time: the budget is the contract, and + // an engine that dispatches within the budget but executes serially has silently turned + // the budget into an upper bound nothing reaches. + constexpr Uint kBurst = 4; // == kTestThreads, so the budget can hold all of them at once + ShaderCompilePool pool(kTestThreads); + + std::atomic live{0}; + std::atomic peak{0}; + std::atomic finished{0}; + + Vector> burst; + burst.reserve(kBurst); + for (Uint i = 0; i < kBurst; ++i) { + burst.push_back(MakeShared([&](TestJob&) { + const Uint now = live.fetch_add(1, std::memory_order_acq_rel) + 1; + Uint seen = peak.load(std::memory_order_acquire); + while (now > seen && !peak.compare_exchange_weak(seen, now, std::memory_order_acq_rel)) { + } + // Long enough that a serial engine cannot fake overlap, short enough to keep the + // case cheap: with any real spread every body is inside this window together. + std::this_thread::sleep_for(std::chrono::milliseconds(120)); + live.fetch_sub(1, std::memory_order_acq_rel); + finished.fetch_add(1, std::memory_order_acq_rel); + })); + } + + std::atomic postedFromPoolThread{false}; + auto seeder = MakeShared([&](TestJob&) { + postedFromPoolThread.store(ShaderCompilePool::IsPoolThread(), std::memory_order_release); + for (const auto& job : burst) pool.Post(job); + }); + + pool.Post(seeder); + seeder->Wait(); + for (const auto& job : burst) job->Wait(); + + ASSERT_TRUE(postedFromPoolThread.load()) << "the burst was not posted from a pool thread"; + EXPECT_EQ(finished.load(), kBurst); + EXPECT_GT(peak.load(), 1u) << "a burst posted from inside the pool ran strictly one at a time; the " + "engine serialized work the budget had already cleared"; +} + TEST(JobNodeSubmit, ConcurrencyBudgetIsNeverExceeded) { constexpr Uint kBudget = 2; constexpr Uint kJobs = 64; @@ -472,30 +631,58 @@ TEST(JobNodeException, AThrowingJobDoesNotPoisonTheWorkerForLaterJobs) { // --------------------------------------------------------------------------------------- TEST(ShaderCompilePoolDrain, StopAndDrainWithAThousandQueuedJobsLeavesNoneRunningOrPending) { - constexpr Uint kJobs = 1000; + constexpr Uint kQueued = 1000; ShaderCompilePool pool(kTestThreads); - pool.SetMaxConcurrency(1); // keep the vast majority queued behind the budget + pool.SetMaxConcurrency(1); // one slot, so everything behind the first job stays queued - Vector> jobs; - jobs.reserve(kJobs); - for (Uint i = 0; i < kJobs; ++i) { - jobs.push_back(MakeShared()); - pool.Post(jobs.back()); + // Pin that slot with a job that will not return until this test says so. Everything + // posted behind it is then PROVABLY still in the queue, which is what makes the counts + // below exact. + // + // This case used to post a thousand trivial jobs and drain immediately, hoping the drain + // would beat the workers to some of them - and then assert only that "some" were + // cancelled. That hope does not survive an engine whose workers take their next job + // without a scheduler round trip: the libfork engine drained all thousand before the + // posting loop had finished, so the assertion failed about one run in fifty. The property + // being tested (a drain ABANDONS queued work rather than running it) is real and + // engine-independent; only the way it was provoked was a race. + Gate gate; + std::atomic entered{false}; + auto blocker = MakeShared([&](TestJob&) { + entered.store(true, std::memory_order_release); + gate.Wait(); + }); + pool.Post(blocker); + ASSERT_TRUE(WaitUntil([&] { return entered.load(); })); + + Vector> queued; + queued.reserve(kQueued); + for (Uint i = 0; i < kQueued; ++i) { + queued.push_back(MakeShared()); + pool.Post(queued.back()); } + for (const auto& job : queued) ASSERT_FALSE(job->IsTerminal()); - pool.StopAndDrain(); + std::thread drain([&] { pool.StopAndDrain(); }); + // StopAndDrain settles the entire queue before it waits for the running body, so the + // first cancelled node proves it is past that point - and the gate can then be released + // without racing it. + ASSERT_TRUE(WaitUntil([&] { return queued.front()->IsTerminal(); })); + gate.Open(); + drain.join(); - // Every node is terminal, so nothing can be waiting on a worker that will never come. - Uint complete = 0; - Uint cancelled = 0; - for (const auto& job : jobs) { + // The job that was already running still finished: an in-flight body is waited for, not + // interrupted. + EXPECT_TRUE(blocker->IsComplete()); + EXPECT_EQ(blocker->ran.load(), 1u); + + // And every queued node is terminal, so nothing is left waiting on a worker that will + // never come - settled as cancelled, with its body never entered. + for (const auto& job : queued) { ASSERT_TRUE(job->IsTerminal()); - if (job->IsComplete()) ++complete; - if (job->IsCancelled()) ++cancelled; - EXPECT_LE(job->ran.load(), 1u); + EXPECT_TRUE(job->IsCancelled()); + EXPECT_EQ(job->ran.load(), 0u); } - EXPECT_EQ(complete + cancelled, kJobs); - EXPECT_GT(cancelled, 0u); // the drain really did abandon queued work rather than run it } TEST(ShaderCompilePoolDrain, StopAndDrainWaitsForARunningBodyToReturn) { @@ -535,3 +722,141 @@ TEST(ShaderCompilePoolDrain, JobsPostedAfterADrainStillRun) { EXPECT_TRUE(job->IsComplete()); EXPECT_EQ(job->ran.load(), 1u); } + +// --------------------------------------------------------------------------------------- +// Execution engine selection (MOBILEGL_ASYNC_POOL) +// --------------------------------------------------------------------------------------- +// +// The engine decides only HOW a job that the concurrency budget has already cleared reaches a +// worker thread. Everything else in this file - the budget, cancel request-vs-outcome, the +// continuation machinery, the inline fallback after a stop, the drain - is engine-independent +// by construction, which is why the whole suite is expected to pass unchanged with +// MOBILEGL_ASYNC_POOL unset and with it set to libfork. These cases pin the selection itself, +// so that a run of the matrix cannot silently test asio twice. + +TEST(AsyncPoolEngineSelection, EveryAcceptedSpellingParsesToItsEngine) { + EXPECT_EQ(ParseAsyncPoolEngine("asio"), AsyncPoolEngine::Asio); + EXPECT_EQ(ParseAsyncPoolEngine("libfork"), AsyncPoolEngine::Libfork); + // Case-insensitive, like the other named-value variables (MOBILEGL_*_MULTIDRAW_MODE). + EXPECT_EQ(ParseAsyncPoolEngine("Libfork"), AsyncPoolEngine::Libfork); + EXPECT_EQ(ParseAsyncPoolEngine("LIBFORK"), AsyncPoolEngine::Libfork); + EXPECT_EQ(ParseAsyncPoolEngine("ASIO"), AsyncPoolEngine::Asio); + + EXPECT_STREQ(AsyncPoolEngineName(AsyncPoolEngine::Asio), "asio"); + EXPECT_STREQ(AsyncPoolEngineName(AsyncPoolEngine::Libfork), "libfork"); + // Round trip: whatever the name prints is a spelling the variable accepts back. + EXPECT_EQ(ParseAsyncPoolEngine(AsyncPoolEngineName(AsyncPoolEngine::Asio)), AsyncPoolEngine::Asio); + EXPECT_EQ(ParseAsyncPoolEngine(AsyncPoolEngineName(AsyncPoolEngine::Libfork)), AsyncPoolEngine::Libfork); +} + +TEST(AsyncPoolEngineSelection, EmptyAndAutoAreTheDefaultEngineAndSaySoSilently) { + // Unset resolves through the empty string, and "auto" is the spelling the other named + // -value variables accept for "no preference". Neither is a mistake, so neither warns. + const std::streamoff before = LogSize(); + EXPECT_EQ(ParseAsyncPoolEngine(""), AsyncPoolEngine::Asio); + EXPECT_EQ(ParseAsyncPoolEngine("auto"), AsyncPoolEngine::Asio); + EXPECT_EQ(ReadLogFrom(before).find("MOBILEGL_ASYNC_POOL"), String::npos) + << "a legitimate value warned; only an unrecognized one may"; +} + +TEST(AsyncPoolEngineSelection, AnUnrecognizedEngineNameFallsBackToAsioAndWarns) { + const std::streamoff before = LogSize(); + EXPECT_EQ(ParseAsyncPoolEngine("libfrok"), AsyncPoolEngine::Asio); + +// The warning is the other half of the contract: a misspelt engine name that fell back +// silently would be indistinguishable from an unset variable, and a scaling measurement +// taken against the wrong engine is worse than no measurement. +// +// Guarded because MGLOG_W is a compile-time no-op unless the build's log level admits it - +// and the shipped level does not (Log.h orders the levels DEBUG=0, WARN=1, ERROR=2, INFO=3, +// FATAL=4 and gates on `ACTIVE <= LEVEL`, so the default INFO build enables only INFO and +// FATAL). Nothing is skipped: the fallback above is pinned in every build, and this half is +// checked by a build configured with +// -DMOBILEGL_LOG_ACTIVE_LEVEL=MOBILEGL_LOG_LEVEL_WARN. The same guard is what makes the +// preceding "says so silently" case honest rather than vacuously true. +#if MOBILEGL_LOG_ACTIVE_LEVEL <= MOBILEGL_LOG_LEVEL_WARN + const String logged = ReadLogFrom(before); + EXPECT_NE(logged.find("MOBILEGL_ASYNC_POOL"), String::npos) << "no warning names the variable; log tail: " << logged; + EXPECT_NE(logged.find("libfrok"), String::npos) + << "the warning does not quote the rejected value; log tail: " << logged; + EXPECT_NE(logged.find("asio"), String::npos) + << "the warning does not say what it fell back to; log tail: " << logged; +#else + (void)before; +#endif +} + +TEST(AsyncPoolEngineSelection, TheDetectedEngineIsTheOneTheEnvironmentAskedFor) { + // Read the variable directly rather than through the pool, so this really compares the + // process's answer against the environment the runner exported. This is the case that + // makes "the suite passed with MOBILEGL_ASYNC_POOL=libfork" mean something. + const char* const raw = std::getenv("MOBILEGL_ASYNC_POOL"); + const AsyncPoolEngine expected = ParseAsyncPoolEngine(raw != nullptr ? String(raw) : String()); + EXPECT_EQ(DetectAsyncPoolEngine(), expected); + + // Stable: resolved once per process, so it cannot drift between calls. + EXPECT_EQ(DetectAsyncPoolEngine(), DetectAsyncPoolEngine()); + + if (DetectAsyncPoolEngine() != AsyncPoolEngine::Asio) { + // Selecting a non-default engine announces itself at INFO, which the shipped log level + // does admit - so on the libfork half of the matrix this doubles as the positive + // control for the log plumbing the preceding two cases read: it proves + // MOBILEGL_LOG_FILE_PATH took effect and that ReadLogFrom really sees MobileGL's + // output, rather than passing because the file is always empty. + const String logged = ReadLogFrom(0); + EXPECT_NE(logged.find("MOBILEGL_ASYNC_POOL"), String::npos) + << "the selected engine was never announced, so this binary's log capture proves nothing"; + EXPECT_NE(logged.find(AsyncPoolEngineName(DetectAsyncPoolEngine())), String::npos); + } +} + +TEST(AsyncPoolEngineSelection, EveryPoolReportsTheProcessEngineAndRunsWorkOnIt) { + ShaderCompilePool first(kTestThreads); + ShaderCompilePool second(kTestThreads); + EXPECT_EQ(first.GetEngine(), DetectAsyncPoolEngine()); + EXPECT_EQ(second.GetEngine(), first.GetEngine()) + << "two pools in one process disagree about the engine; a process must never run both"; + + // And the engine it reports is the one that actually executed the work: the body ran off + // the calling thread, on a thread the pool owns. + const auto callingThread = std::this_thread::get_id(); + std::atomic sawPoolThread{false}; + std::thread::id bodyThread{}; + auto job = MakeShared([&](TestJob&) { + sawPoolThread.store(ShaderCompilePool::IsPoolThread(), std::memory_order_release); + bodyThread = std::this_thread::get_id(); + }); + first.Post(job); + job->Wait(); + + ASSERT_TRUE(job->IsComplete()); + EXPECT_TRUE(sawPoolThread.load()); + EXPECT_NE(bodyThread, callingThread); +} + +// gtest_main is replaced here for one reason: the engine-selection cases above assert that an +// unrecognized MOBILEGL_ASYNC_POOL value WARNS, and MobileGL's desktop log sink is the log +// file - MOBILEGL_LOG_ENABLE_CONSOLE is 0 in Defines.h, so there is nothing on stdout to +// capture. MOBILEGL_LOG_FILE_PATH is read by Log.cpp's InitFile() at the first log write in +// the process, so it has to be set before any test body runs. +int main(int argc, char** argv) { + const std::filesystem::path logPath = + std::filesystem::temp_directory_path() / + ("mobilegl-jobnodetest-" + std::to_string(static_cast(MGL_TEST_GETPID())) + ".log"); + g_logFilePath = logPath.string(); + std::filesystem::remove(logPath); +#ifdef _WIN32 + ::_putenv_s("MOBILEGL_LOG_FILE_PATH", g_logFilePath.c_str()); +#else + ::setenv("MOBILEGL_LOG_FILE_PATH", g_logFilePath.c_str(), 1); +#endif + + ::testing::InitGoogleTest(&argc, argv); + const int result = RUN_ALL_TESTS(); + + // Best-effort: leaving a log file per test process in the temp directory would be litter, + // and a failed run has already printed the tail it needed into the gtest output. + std::error_code ignored; + std::filesystem::remove(logPath, ignored); + return result; +} diff --git a/MobileGL/MG_Util/Async/ShaderCompilePool.cpp b/MobileGL/MG_Util/Async/ShaderCompilePool.cpp index c4953e57..dc1d1219 100644 --- a/MobileGL/MG_Util/Async/ShaderCompilePool.cpp +++ b/MobileGL/MG_Util/Async/ShaderCompilePool.cpp @@ -12,8 +12,14 @@ #include #include +#include +#include + #include +#include #include +#include +#include namespace MobileGL::MG_Util::Async { namespace { @@ -135,46 +141,506 @@ namespace MobileGL::MG_Util::Async { return std::clamp(DetectBigCoreCount(), 1u, kMaxAutoShaderCompileThreads); } + // ---- Engine selection ----------------------------------------------------------------- + + const char* AsyncPoolEngineName(const AsyncPoolEngine engine) { + switch (engine) { + case AsyncPoolEngine::Libfork: return "libfork"; + case AsyncPoolEngine::Asio: break; + } + return "asio"; + } + + AsyncPoolEngine ParseAsyncPoolEngine(const String& value) { + String lowered = value; + std::transform(lowered.begin(), lowered.end(), lowered.begin(), + [](const unsigned char c) { return static_cast(std::tolower(c)); }); + if (lowered == "libfork") return AsyncPoolEngine::Libfork; + if (lowered == "asio" || lowered == "auto" || lowered.empty()) return AsyncPoolEngine::Asio; + // Not silent: a misspelt engine name resolving to the default would be + // indistinguishable from not having set the variable at all, and the only reason to + // set it is to know which engine ran. + MGLOG_W("Config: Ignoring invalid env variable MOBILEGL_ASYNC_POOL='%s'; expected asio|libfork, " + "using asio", + value.c_str()); + return AsyncPoolEngine::Asio; + } + + AsyncPoolEngine DetectAsyncPoolEngine() { + // A live std::getenv rather than an MG_Config::Features mirror, and deliberately so: + // a ShaderCompilePool is constructed by binaries that never call MobileGL::Initialize() + // and therefore never run MG_ConfigLoader::Init() - MG_Test/Util/JobNodeTest builds + // pools directly, and it is the suite that exercises the engines against each other. + // Reading Features there would silently resolve to the default and the libfork half of + // the test matrix would prove nothing. See the exemption list in Config.h. + // + // Resolved once per process (a function-local static): every pool in a process gets + // the same engine, so a process can never end up running two. + static const AsyncPoolEngine engine = [] { + const char* value = std::getenv("MOBILEGL_ASYNC_POOL"); + const AsyncPoolEngine resolved = ParseAsyncPoolEngine(value != nullptr ? String(value) : String()); + if (resolved != AsyncPoolEngine::Asio) { + MGLOG_I("ShaderCompilePool: MOBILEGL_ASYNC_POOL selected the %s execution engine", + AsyncPoolEngineName(resolved)); + } + return resolved; + }(); + return engine; + } + + namespace { + // ---- The engine boundary ---------------------------------------------------------- + // Submit() has exactly asio::post's contract, and ShaderCompilePool::Impl leans on all + // four halves of it: + // * it NEVER runs `fn` on the calling thread. DispatchLocked calls it while holding + // the pool's plain, non-recursive mutex, and a job body (or a terminal + // continuation it releases) is free to call Post() again - an inline run would + // deadlock on the lock this frame already owns. + // * it is callable from ANY thread, a worker of this very pool included: + // ProgramLinkTask::OnDepSettled posts the link job from whichever thread drove the + // last compile terminal, which is a worker. + // * it may throw, and when it does it must not have consumed the caller's job node, + // so Post/DispatchLocked can settle the node instead of stranding it Pending with + // a joiner blocked forever. + // * once it has accepted `fn`, `fn` WILL run. A dropped callable is a node nothing + // ever settles, so the engines run it themselves rather than discard it. + class JobExecutor { + public: + virtual ~JobExecutor() = default; + JobExecutor() = default; + JobExecutor(const JobExecutor&) = delete; + JobExecutor& operator=(const JobExecutor&) = delete; + + virtual void Submit(std::function fn) = 0; + + // Returns once every callable ever handed to Submit has finished running. The + // guarantee StopAndDrain sells to library teardown: after it returns, no worker is + // still inside a job body that could touch glslang's process globals. + virtual void JoinAll() = 0; + }; + + // ---- Engine 1: Asio (the shipped default) ----------------------------------------- + class AsioJobExecutor final : public JobExecutor { + public: + explicit AsioJobExecutor(const Uint threads) : m_pool(threads) {} + + // asio::post only enqueues; it never runs the handler on the calling thread, which + // is what makes calling it under the pool mutex safe. + void Submit(std::function fn) override { asio::post(m_pool, Move(fn)); } + + void JoinAll() override { m_pool.join(); } + + private: + asio::thread_pool m_pool; + }; + + // ---- Engine 2: libfork ------------------------------------------------------------ + // + // libfork is a continuation-stealing fork-join runtime, and the shape that fits here is + // NOT fork-join: a job body is one coarse, blocking, non-forking unit (a glslang + // compile), and the concurrency budget that bounds peak RSS is Impl's, not the + // scheduler's. So libfork is used as a job executor - each dispatched job is a detached + // root task - and what it is being asked to beat is Asio's single scheduler queue with + // its per-worker work-stealing deques and sleeping workers. + // + // The one thing libfork forbids is the thing this pool does constantly: lf::schedule + // (which lf::detach is built on) THROWS lf::schedule_in_worker when the calling thread + // is a libfork worker, because workers may never block. Yet a worker submits on every + // job completion - RunOnWorker's tail refills the budget - and again whenever a + // terminal continuation posts (ProgramLinkTask::OnDepSettled). Routing those through a + // separate dispatch thread works but costs two thread wakeups per job, which measured + // 4x worse than Asio on short jobs. So instead a dispatched root is a CHAIN: when its + // body returns it takes the next queued job itself and runs it in the same coroutine + // on the same worker. The refill a worker submits is therefore absorbed by the very + // chain that submitted it - no scheduler round trip, no wakeup - and libfork is only + // entered for work that arrives from outside the pool. + // + // Absorption is bounded at one job per running chain, though, because a chain is one + // worker: past that bound the queue would be jobs the budget has already cleared, + // waiting behind each other on a single thread. See Submit. + // + // Why none of this can strand a job: the queue below is only ever added to from inside + // a running chain (tl_chainOwner == this), and a chain exits only when it finds the + // queue empty - unconditionally, whatever the bound says. Every other submitter goes + // to the dispatch thread or straight to lf::detach. + class LibforkJobExecutor; + + // Which executor's chain, if any, is running on this thread. Deliberately narrower + // than ShaderCompilePool::IsPoolThread(): that flag is process-wide and latched + // forever, so a worker of a DIFFERENT pool would read as "mine" and queue a job into a + // chain that will never drain it. This says exactly "a chain of *this* executor is + // executing on this thread, and it will look at the queue again before it exits". + thread_local LibforkJobExecutor* tl_chainOwner = nullptr; + + // One dispatched job, heap-owned. It reaches its coroutine as a POINTER passed BY + // VALUE: libfork forwards a root task's arguments into the coroutine frame, so a + // by-value pointer is copied into the frame, whereas anything passed by reference + // would dangle the moment lf::detach returns - and detach, unlike sync_wait, does not + // outlive the task. + struct LibforkJob { + std::function body; + LibforkJobExecutor* owner; + }; + + // A scheduler adaptor for lf::detach: it places external submissions round-robin over + // lf::lazy_pool's worker contexts instead of letting the pool pick one at random. + // Both reasons are load-bearing, and the second was worth 1.3x at a budget equal to + // the worker count - the configuration MobileGL actually ships, since maxConcurrency + // is clamped to the thread count: + // * lf::lazy_pool::schedule chooses its victim with a + // std::uniform_int_distribution over a lazy_pool-member xoshiro generator - + // unsynchronized mutable state, so two concurrent submissions are a data race + // inside libfork itself. An atomic cursor is not. + // * A worker's SUBMISSION list is drained only by that worker + // (worker_context::try_pop_all is documented "for use only by the owning worker + // thread"); a thief takes from the task deque, which is a different queue. So a + // job placed on a worker that is inside a long blocking body waits for that body + // rather than being stolen - and random placement of `budget` submissions over + // `budget` workers collides by the birthday rule. Round-robin lands the GL + // thread's burst one per worker, which is exactly the intended shape. + struct RoundRobinSubmitter { + std::span contexts; + std::atomic* cursor; + + void schedule(const lf::submit_handle job) const { + const Uint64 index = cursor->fetch_add(1, std::memory_order_relaxed); + contexts[static_cast(index % contexts.size())]->schedule(job); + } + }; + + void RunLibforkChain(LibforkJob* raw) noexcept; + + // The root task every dispatched chain runs as. libfork async function objects are + // copyable, captureless callables returning lf::task<>, whose first parameter is the + // combinator's synthesized first argument (unused here: this task neither forks nor + // joins). The coroutine exists purely as libfork's entry protocol; the loop is in + // RunLibforkChain. + inline constexpr auto kLibforkChainTask = [](auto /*self*/, LibforkJob* job) -> lf::task { + RunLibforkChain(job); + co_return; + }; + + class LibforkJobExecutor final : public JobExecutor { + public: + explicit LibforkJobExecutor(const Uint threads) + : m_pool(static_cast(std::max(1u, threads))), m_contexts(m_pool.contexts()), + m_fallback([this] { FallbackLoop(); }) {} + + ~LibforkJobExecutor() override { + JoinAll(); + { + const std::lock_guard lock(m_mutex); + m_fallbackStop = true; + } + m_fallbackCv.notify_all(); + if (m_fallback.joinable()) m_fallback.join(); + // m_pool is destroyed last, and only here: lf::lazy_pool may not be destructed + // while any submitted task can still run or submit more. JoinAll() has + // established the first and the joined fallback thread the second. Its + // destructor then joins the worker threads, so a worker still unwinding a + // finished coroutine frame is waited for rather than pulled out from under. + } + + void Submit(std::function fn) override { + if (tl_chainOwner == this) { + const std::lock_guard lock(m_mutex); + // The hot path: ONE job per running chain. A chain picks up exactly one + // queued job each time its body returns, so a queue no longer than the + // number of live chains is a queue every entry of which has a distinct + // worker waiting to take it - which is precisely the steady state this + // absorption exists for (every worker finishes a job and refills its own + // slot, all at once, with no scheduler round trip between them). + // + // Past that it is oversubscription, and absorbing it would be a + // correctness-preserving way to destroy the pool's parallelism: the + // budget would still say `maxConcurrency` jobs are in flight while one + // worker ran them one behind another. That is not hypothetical - it is + // the tail of a pack load, where one compile going terminal releases + // several programs at once (ShaderCompileAdoptionMap lets a single + // compile settle many) and the worker that drove it posts the whole + // burst into an otherwise idle pool. Measured before this branch existed: + // four such jobs took 4x one job's wall time on libfork and 1x on Asio. + // + // The overflow cannot go to lf::detach from here - a libfork worker may + // not schedule - so it goes to the dispatch thread, which detaches it to + // a worker of its own. That costs one thread wakeup; serializing costs a + // whole compile. + // + // The count is taken AFTER the push, not before: deque::push_back is + // strongly exception-safe, so an allocation failure here leaves `fn` + // intact for DispatchLocked to settle - but a count incremented in front + // of it would be a count nothing ever gives back, and JoinAll would wait + // on it forever. + const Bool takeable = m_chainQueue.size() < m_liveChains; + if (takeable) { + m_chainQueue.push_back(Move(fn)); + ++m_outstanding; + } else { + m_fallbackQueue.push_back(Move(fn)); + ++m_outstanding; + m_fallbackCv.notify_one(); + } + return; + } + + { + // Counted before anything can run it, so JoinAll cannot observe a zero + // that this job would have broken. + const std::lock_guard lock(m_mutex); + ++m_outstanding; + } + try { + DetachChain(Move(fn)); + } catch (const lf::schedule_in_worker&) { + // Submitted from a libfork worker that is not running one of my chains - + // a worker of another ShaderCompilePool. libfork will not take a + // submission from there at all, and the queue above is not safe for it + // (no chain of mine is running on that thread to drain it), so it goes to + // the fallback thread, which is neither. DetachChain restored `fn` before + // it threw. + const std::lock_guard lock(m_mutex); + m_fallbackQueue.push_back(Move(fn)); + m_fallbackCv.notify_one(); + } catch (...) { + // Out of memory. Give the count back and let the caller settle its node: + // that is Submit's contract and what DispatchLocked is written against. + Retire(); + throw; + } + } + + void JoinAll() override { + std::unique_lock lock(m_mutex); + m_idleCv.wait(lock, [this] { return m_outstanding == 0; }); + } + + // A chain announces itself before it runs its first body, so that Submit's + // absorption rule can count the workers that are going to come back and ask for + // more. Under-counting is the only direction this can be wrong in (a detached + // chain is not counted until it starts), and under-counting only sends work to + // the dispatch thread that a chain could have taken - never the reverse. + void EnterChain() noexcept { + const std::lock_guard lock(m_mutex); + ++m_liveChains; + } + + // The end of one job in a chain. Returns true having loaded `body` with the next + // job to run on this same worker, false when there is nothing left - after which + // the caller must touch neither `this` nor anything owned by it, because the + // count this drops to zero may be the one JoinAll is waiting for. + // + // `body` must arrive empty: the finished job's captures (a strong reference to its + // JobNode) are released by the chain, outside this lock, so that no JobNode + // destructor ever runs inside the executor's critical section. + Bool RetireAndTakeNext(std::function& body) noexcept { + const std::lock_guard lock(m_mutex); + --m_outstanding; + if (!m_chainQueue.empty()) { + // Unconditional, and it has to stay that way: a chain that exited while + // the queue was non-empty could be the last one, and the entry would then + // be waiting on a worker that never comes. That is what makes the + // absorption bound in Submit a scheduling policy rather than a liveness + // requirement. + // + // swap, not move-assign: std::function's move assignment is not noexcept, + // and this function is. + body.swap(m_chainQueue.front()); + m_chainQueue.pop_front(); + return true; // the taken job's own count stays held + } + --m_liveChains; + // Notified while STILL HOLDING the lock, which is the whole reason this is not + // the usual notify-after-unlock. The wakeup this sends can be the one that + // lets JoinAll return and ~LibforkJobExecutor destroy m_idleCv - and a + // std::condition_variable may not be destroyed while another thread is inside + // notify_all() on it. Holding the lock across the notify means the waiter + // cannot re-acquire the mutex, and therefore cannot leave wait(), until this + // thread is out of both the notify and the unlock. ThreadSanitizer catches the + // other order immediately (pthread_cond_destroy vs pthread_cond_broadcast). + if (m_outstanding == 0) m_idleCv.notify_all(); + return false; + } + + private: + // Builds the root task and hands it to libfork. On any failure `fn` is restored, + // so the caller can still decide what to do with the job. + void DetachChain(std::function&& fn) { + // `new T{...}` allocates before it constructs, so a throwing operator new + // leaves `fn` untouched; the member move is std::function's noexcept one. + LibforkJob* job = new LibforkJob{Move(fn), this}; + try { + lf::detach(RoundRobinSubmitter{m_contexts, &m_cursor}, kLibforkChainTask, job); + } catch (...) { + // lf::schedule upholds the strong exception guarantee, so nothing was + // scheduled and the payload is still ours. + const UniquePtr owned(job); + fn = Move(owned->body); + throw; + } + } + + void Retire() noexcept { + // Under the lock, for the reason RetireAndTakeNext spells out. + const std::lock_guard lock(m_mutex); + if (--m_outstanding == 0) m_idleCv.notify_all(); + } + + // The dispatch thread. It exists because lf::detach is illegal on a libfork worker + // and legal here, and it serves the two cases Submit cannot take itself: a + // submission from another pool's worker, and a chain's overflow past the + // one-job-per-chain bound. It sleeps otherwise, and it dispatches rather than + // executes - a body only ever runs here if libfork refuses the job outright. + void FallbackLoop() { + for (;;) { + std::function fn; + { + std::unique_lock lock(m_mutex); + m_fallbackCv.wait(lock, [this] { return !m_fallbackQueue.empty() || m_fallbackStop; }); + // Emptiness is checked before the stop flag so that a stop can never + // strand accepted work: an accepted job always runs, because the node + // behind it has a joiner that would otherwise block forever. + if (m_fallbackQueue.empty()) return; + fn.swap(m_fallbackQueue.front()); + m_fallbackQueue.pop_front(); + } + try { + DetachChain(Move(fn)); + } catch (...) { + MGLOG_E("ShaderCompilePool: libfork refused a fallback dispatch; running the job on " + "the dispatch thread instead of dropping it"); + RunHere(Move(fn)); + } + } + } + + // Last resort. Running the body here costs this engine its parallelism for one + // job; dropping it would cost a joiner its wakeup forever. + void RunHere(std::function&& fn) noexcept { + try { + if (fn) fn(); + } catch (...) { + MGLOG_E("ShaderCompilePool: a job body escaped its own containment on the dispatch " + "thread; it has been swallowed to keep the thread alive"); + } + fn = nullptr; + Retire(); + } + + lf::lazy_pool m_pool; + // Fixed for the pool's lifetime, so it is read once rather than per submission. + std::span m_contexts; + std::atomic m_cursor{0}; + + std::mutex m_mutex; + std::condition_variable m_fallbackCv; + std::condition_variable m_idleCv; + // Refills and continuations submitted from inside a chain: drained by the chains. + std::deque> m_chainQueue; + // Chains currently executing, i.e. workers that will look at m_chainQueue again + // before they exit. The bound on how much Submit may absorb into a chain. + Uint m_liveChains = 0; + // Submissions from another pool's libfork worker, and the overflow of the rule + // above: drained by m_fallback, which detaches each one to a worker. + std::deque> m_fallbackQueue; + // Everything submitted and not yet finished, whichever queue it is in and whether + // or not it has reached a worker, so JoinAll needs a single predicate. + Uint m_outstanding = 0; + Bool m_fallbackStop = false; + std::thread m_fallback; + }; + + void RunLibforkChain(LibforkJob* const raw) noexcept { + UniquePtr job(raw); + LibforkJobExecutor* const owner = job->owner; + std::function body; + body.swap(job->body); + job.reset(); + + LibforkJobExecutor* const savedOwner = tl_chainOwner; + tl_chainOwner = owner; + owner->EnterChain(); + + for (;;) { + try { + if (body) body(); + } catch (...) { + // JobNode::Run contains every body exception already; this is the backstop + // for the wrapper itself. An exception escaping here would be stashed in + // the root task's shared state, which lf::detach discards - i.e. silently + // lost - and would abandon the rest of the chain. + MGLOG_E("ShaderCompilePool: a job body escaped its own containment on a libfork worker; " + "it has been swallowed to keep the chain alive"); + } + // Release the finished job's captures (its strong JobNode reference) HERE, + // outside the executor's lock: a JobNode destructor is arbitrary code. + body = nullptr; + if (!owner->RetireAndTakeNext(body)) break; + } + + // `owner` may already be destroyed - RetireAndTakeNext returning false can be the + // call that releases a JoinAll. Nothing below touches it. + tl_chainOwner = savedOwner; + } + + UniquePtr MakeJobExecutor(const AsyncPoolEngine engine, const Uint threads) { + switch (engine) { + case AsyncPoolEngine::Libfork: return MakeUnique(threads); + case AsyncPoolEngine::Asio: break; + } + return MakeUnique(threads); + } + } // namespace + struct ShaderCompilePool::Impl { - explicit Impl(const Uint threads) : threadCount(std::max(1u, threads)), maxConcurrency(threadCount) {} + explicit Impl(const Uint threads) + : threadCount(std::max(1u, threads)), engine(DetectAsyncPoolEngine()), maxConcurrency(threadCount) {} const Uint threadCount; + // Latched at construction, not re-read: a pool may not change engines under its own + // workers, and GetEngine() is what the tests compare against the environment. + const AsyncPoolEngine engine; std::mutex mutex; - // Created on the first dispatched Post, never in the constructor: asio::thread_pool - // spawns its threads eagerly, and a build with async off must not pay for threads it + // Created on the first dispatched Post, never in the constructor: both engines spawn + // their threads eagerly (asio::thread_pool its workers, lf::lazy_pool its workers plus + // this file's dispatch thread), and a build with async off must not pay for threads it // will never use. - UniquePtr pool; + UniquePtr executor; std::deque> queue; Uint inFlight = 0; Uint maxConcurrency; std::atomic stopped{false}; - // Callers hold `mutex`. Hands as many queued nodes to Asio as the concurrency budget - // allows. Posting under the lock is safe and is what keeps `pool` from being moved - // out by a concurrent StopAndDrain between the decision and the dispatch: asio::post - // only enqueues, it never runs the handler on the calling thread, so it cannot + // Callers hold `mutex`. Hands as many queued nodes to the engine as the concurrency + // budget allows. Submitting under the lock is safe and is what keeps `executor` from + // being moved out by a concurrent StopAndDrain between the decision and the dispatch: + // Submit only enqueues, it never runs the callable on the calling thread, so it cannot // re-enter this mutex. // - // A node asio::post fails to hand off is appended to `toCancel` instead of being + // A node the engine fails to accept is appended to `toCancel` instead of being // Cancel()'d here: Cancel() runs the node's OnTerminal continuations inline (stage 4 // added ProgramLinkTask::OnDepSettled as a real one), and a continuation is free to // call ShaderCompilePool::Post() again. Every caller of DispatchLocked holds `mutex` // (a plain, non-recursive std::mutex) - Cancel()'ing in here would let that // re-entrant Post() deadlock on the very lock this frame already owns. The caller // drains `toCancel` after releasing the lock. + // + // The `stopped` check is also what keeps this loop from dereferencing a null + // `executor`: StopAndDrain sets the flag and moves the executor out in the same + // critical section, so a stopped pool never reaches the Submit below. void DispatchLocked(Vector>& toCancel) { while (!queue.empty() && inFlight < maxConcurrency && !stopped.load(std::memory_order_acquire)) { - // Copy rather than move into the handler: if asio::post throws (it allocates) - // the local SharedPtr is still valid, so the node can be settled instead of - // being stranded Pending in a queue nothing will dispatch from again - a - // joiner would block on it forever. Reclaiming the slot matters just as much: - // a leaked `inFlight` shrinks the pool's concurrency budget permanently. + // Copy rather than move into the callable: if Submit throws (both engines + // allocate) the local SharedPtr is still valid, so the node can be settled + // instead of being stranded Pending in a queue nothing will dispatch from + // again - a joiner would block on it forever. Reclaiming the slot matters just + // as much: a leaked `inFlight` shrinks the pool's concurrency budget + // permanently. SharedPtr node = queue.front(); queue.pop_front(); ++inFlight; try { - asio::post(*pool, [this, node]() mutable { RunOnWorker(Move(node)); }); + executor->Submit([this, node]() mutable { RunOnWorker(Move(node)); }); } catch (...) { --inFlight; toCancel.push_back(Move(node)); @@ -184,7 +650,7 @@ namespace MobileGL::MG_Util::Async { void RunOnWorker(SharedPtr node) { tl_isPoolThread = true; - // A node that was already handed to Asio when StopAndDrain ran still arrives + // A node that was already handed to the engine when StopAndDrain ran still arrives // here; cancelling it first turns the dispatch into a state transition instead of // a full compile, so the drain's join() returns promptly. This Cancel() runs // before `mutex` is ever taken in this frame, so it is not subject to the @@ -234,13 +700,15 @@ namespace MobileGL::MG_Util::Async { return m_impl->maxConcurrency; } + AsyncPoolEngine ShaderCompilePool::GetEngine() const { return m_impl->engine; } + void ShaderCompilePool::SetMaxConcurrency(const Uint n) { Vector> toCancel; { const std::lock_guard lock(m_impl->mutex); m_impl->maxConcurrency = std::clamp(n, 1u, m_impl->threadCount); // Raising the budget releases whatever the old one was holding back. - if (m_impl->pool) m_impl->DispatchLocked(toCancel); + if (m_impl->executor) m_impl->DispatchLocked(toCancel); } // Outside the lock: see DispatchLocked's comment. for (const auto& n2 : toCancel) { @@ -252,23 +720,23 @@ namespace MobileGL::MG_Util::Async { if (!node) return; EnsureProcessTeardownSentinel(); - // Enqueueing can throw: the thread_pool construction and asio::post both allocate, - // and under memory pressure a throw here would escape glCompileShader leaving the - // node Pending with nothing left to dispatch it - the first observable read would - // then block the GL thread forever. Settle the node instead: a cancelled node is a - // state every joiner already handles. + // Enqueueing can throw: building the engine and submitting to it both allocate (and + // both spawn threads), and under memory pressure a throw here would escape + // glCompileShader leaving the node Pending with nothing left to dispatch it - the + // first observable read would then block the GL thread forever. Settle the node + // instead: a cancelled node is a state every joiner already handles. // - // `node` is still valid in the catch for every throw this try can produce. The - // thread_pool construction runs before the move; deque::push_back is strongly - // exception-safe and SharedPtr's move constructor is noexcept, so a throwing - // push_back never consumed it; and DispatchLocked contains its own asio::post - // failures rather than propagating them (see above). Keep it that way. + // `node` is still valid in the catch for every throw this try can produce. The engine + // construction runs before the move; deque::push_back is strongly exception-safe and + // SharedPtr's move constructor is noexcept, so a throwing push_back never consumed it; + // and DispatchLocked contains its own Submit failures rather than propagating them + // (see above). Keep it that way. Bool enqueued = false; Vector> toCancel; try { const std::lock_guard lock(m_impl->mutex); if (!m_impl->stopped.load(std::memory_order_acquire) && !InProcessTeardown()) { - if (!m_impl->pool) m_impl->pool = MakeUnique(m_impl->threadCount); + if (!m_impl->executor) m_impl->executor = MakeJobExecutor(m_impl->engine, m_impl->threadCount); m_impl->queue.push_back(Move(node)); m_impl->DispatchLocked(toCancel); enqueued = true; @@ -304,17 +772,17 @@ namespace MobileGL::MG_Util::Async { } void ShaderCompilePool::StopAndDrain() { - // asio::thread_pool::join() from a pool thread would deadlock on itself, and the - // whole point of this call is that the GL thread waits for the workers. + // Waiting for the workers from a worker would deadlock on itself (asio's join() says + // so outright), and the whole point of this call is that the GL thread waits. MOBILEGL_ASSERT(!IsPoolThread(), "ShaderCompilePool::StopAndDrain() called from a pool thread"); std::deque> abandoned; - UniquePtr pool; + UniquePtr executor; { const std::lock_guard lock(m_impl->mutex); m_impl->stopped.store(true, std::memory_order_release); abandoned.swap(m_impl->queue); - pool = Move(m_impl->pool); + executor = Move(m_impl->executor); } // Queued but never dispatched: settle them so anything chained behind them is @@ -323,9 +791,9 @@ namespace MobileGL::MG_Util::Async { if (node) node->Cancel(); } - if (pool) { - pool->join(); // returns once every handler already handed to Asio has finished - pool.reset(); + if (executor) { + executor->JoinAll(); // returns once every job already handed to the engine is done + executor.reset(); // and this stops the engine's threads } const std::lock_guard lock(m_impl->mutex); diff --git a/MobileGL/MG_Util/Async/ShaderCompilePool.h b/MobileGL/MG_Util/Async/ShaderCompilePool.h index 288377f6..9e9f8a45 100644 --- a/MobileGL/MG_Util/Async/ShaderCompilePool.h +++ b/MobileGL/MG_Util/Async/ShaderCompilePool.h @@ -11,11 +11,12 @@ #include #include -// This header deliberately includes NO Asio header: asio::thread_pool lives behind the pimpl -// in ShaderCompilePool.cpp. Asio stays a private implementation detail of one translation -// unit, so no consumer target (MG_Test, MG_IntegrationTest, MG_Benchmark - each with its own -// target_include_directories) needs the Asio include path, and no consumer pays its compile -// time. Do not add one here. +// This header deliberately includes NO Asio and NO libfork header: both execution engines +// live behind the pimpl in ShaderCompilePool.cpp. They stay private implementation details of +// one translation unit, so no consumer target (MG_Test, MG_IntegrationTest, MG_Benchmark - +// each with its own target_include_directories) needs either include path, and no consumer +// pays their compile time. libfork in particular is a C++20-coroutine header set whose +// instantiation cost nothing outside the pool has any reason to carry. Do not add one here. namespace MobileGL::MG_Util::Async { // Stage 7: on by default. The gate behind the flip (2026-08-09, headless Mesa, both @@ -66,6 +67,31 @@ namespace MobileGL::MG_Util::Async { // MOBILEGL_ASYNC_SHADER_COMPILE_THREADS overrides it outright. Uint DetectShaderCompileThreadCount(); + // ---- MOBILEGL_ASYNC_POOL: which engine drives the worker threads ---------------------- + // The engine is ONLY the execution engine. The job queue, the concurrency budget and its + // clamping, the suspension latch, cancel request-vs-outcome, the stopped-is-synchronous + // fallback and the drain are all engine-independent - they live in ShaderCompilePool::Impl + // and are shared verbatim by both engines, which is what lets the whole async suite run + // unchanged against either one. An engine answers exactly one question: how does a job + // that the budget has already cleared reach a worker thread? + enum class AsyncPoolEngine : Uint8 { + Asio, // asio::thread_pool: one shared queue behind Asio's scheduler lock + Libfork, // lf::lazy_pool: per-worker work-stealing deques, workers sleep when idle + }; + + // "asio" / "libfork" - the spelling the environment variable accepts and the log prints. + const char* AsyncPoolEngineName(AsyncPoolEngine engine); + + // Parses one MOBILEGL_ASYNC_POOL value. Case-insensitive; empty, "auto" and anything + // unrecognized resolve to Asio, and an unrecognized value warns (a misspelt engine name + // would otherwise be indistinguishable from the default, and the whole point of the + // variable is to know which engine ran). + AsyncPoolEngine ParseAsyncPoolEngine(const String& value); + + // The process's engine, resolved from MOBILEGL_ASYNC_POOL on first call and cached. Every + // pool constructed afterwards reports the same answer, so a process never mixes engines. + AsyncPoolEngine DetectAsyncPoolEngine(); + class ShaderCompilePool { public: explicit ShaderCompilePool(Uint threadCount); @@ -97,6 +123,11 @@ namespace MobileGL::MG_Util::Async { Uint GetThreadCount() const; Uint GetMaxConcurrency() const; + // The engine this pool was built with, latched at construction from + // DetectAsyncPoolEngine(). Reported rather than re-resolved so that a pool cannot + // change engines under its own workers. + AsyncPoolEngine GetEngine() const; + // Bounded concurrency doubles as the memory bound, and is how // glMaxShaderCompilerThreadsKHR(n) is honoured: a 300-program pack load cannot put // 300 glslang arenas in flight at once. Clamped to [1, thread count]. diff --git a/MobileGL/MG_Util/SelfTest/DriverPost.cpp b/MobileGL/MG_Util/SelfTest/DriverPost.cpp index 5bd3d13a..ca95ded7 100644 --- a/MobileGL/MG_Util/SelfTest/DriverPost.cpp +++ b/MobileGL/MG_Util/SelfTest/DriverPost.cpp @@ -151,12 +151,20 @@ namespace MobileGL::MG_Util::SelfTest { return; } const Uint threads = MG_Util::Async::DetectShaderCompileThreadCount(); + // The execution engine is named here too. It changes no observable GL behaviour - + // both engines run the same job queue under the same budget - but when a scaling + // or stall report comes back from a device, "which engine was this?" is the first + // question, and a POST page is the one artefact that always accompanies it. + const char* const engineName = + MG_Util::Async::AsyncPoolEngineName(MG_Util::Async::DetectAsyncPoolEngine()); builder.Pass(rowName, - format("on with {} compiler thread{}; GL_KHR_parallel_shader_compile is advertised " + format("on with {} compiler thread{} on the {} execution engine; " + "GL_KHR_parallel_shader_compile is advertised " "and GL_MAX_SHADER_COMPILER_THREADS_KHR = {} (set environment variable " - "MOBILEGL_ASYNC_SHADER_COMPILE=0 to disable it, or " - "MOBILEGL_ASYNC_SHADER_COMPILE_THREADS=n to change the count)", - threads, threads == 1 ? "" : "s", threads)); + "MOBILEGL_ASYNC_SHADER_COMPILE=0 to disable it, " + "MOBILEGL_ASYNC_SHADER_COMPILE_THREADS=n to change the count, or " + "MOBILEGL_ASYNC_POOL=asio|libfork to change the engine)", + threads, threads == 1 ? "" : "s", engineName, threads)); } // Appends the four "MobileGL reported ..." rows for one backend section.