mirror of
https://github.com/MobileGL-Dev/MobileGL
synced 2026-09-08 04:08:32 +09:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3025284a6e |
@@ -34,3 +34,6 @@
|
|||||||
[submodule "3rdparty/asio"]
|
[submodule "3rdparty/asio"]
|
||||||
path = 3rdparty/asio
|
path = 3rdparty/asio
|
||||||
url = https://github.com/chriskohlhoff/asio.git
|
url = https://github.com/chriskohlhoff/asio.git
|
||||||
|
[submodule "3rdparty/libfork"]
|
||||||
|
path = 3rdparty/libfork
|
||||||
|
url = https://github.com/ConorWilliams/libfork.git
|
||||||
|
|||||||
+1
Submodule 3rdparty/libfork added at 9b2b844a5f
@@ -373,6 +373,13 @@ set(MOBILEGL_INCLUDE_DIR
|
|||||||
# MG_Util/Async/ShaderCompilePool.cpp includes it, and it stays behind that file's
|
# MG_Util/Async/ShaderCompilePool.cpp includes it, and it stays behind that file's
|
||||||
# pimpl so no consumer target needs this path.
|
# pimpl so no consumer target needs this path.
|
||||||
${CMAKE_SOURCE_DIR}/3rdparty/asio/asio/include
|
${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
|
add_library(${CMAKE_PROJECT_NAME} SHARED
|
||||||
|
|||||||
@@ -66,6 +66,12 @@ namespace MobileGL::MG_Config {
|
|||||||
// - DISPLAY: X11 session variable, not MobileGL configuration.
|
// - DISPLAY: X11 session variable, not MobileGL configuration.
|
||||||
// - MOBILEGL_LOG_FILE_PATH: log-file init runs before MG_ConfigLoader::Init
|
// - MOBILEGL_LOG_FILE_PATH: log-file init runs before MG_ConfigLoader::Init
|
||||||
// (see MG_Util/Debug/Log.cpp).
|
// (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 {
|
struct FeaturesTable {
|
||||||
// MOBILEGL_DISABLE_TIMERQUERY: do not advertise or use GPU timer queries.
|
// MOBILEGL_DISABLE_TIMERQUERY: do not advertise or use GPU timer queries.
|
||||||
Bool DisableTimerQuery = false;
|
Bool DisableTimerQuery = false;
|
||||||
|
|||||||
@@ -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 <algorithm>
|
||||||
|
#include <atomic>
|
||||||
|
#include <chrono>
|
||||||
|
#include <cstdio>
|
||||||
|
#include <cstdlib>
|
||||||
|
#include <cstring>
|
||||||
|
#include <filesystem>
|
||||||
|
#include <fstream>
|
||||||
|
#include <sstream>
|
||||||
|
#include <string>
|
||||||
|
#include <vector>
|
||||||
|
|
||||||
|
#include "Includes.h"
|
||||||
|
#include "Init.h"
|
||||||
|
#include <Config.h>
|
||||||
|
|
||||||
|
#include <MG_Impl/GLImpl/Program/GL_Program.h>
|
||||||
|
#include <MG_Util/Async/JobNode.h>
|
||||||
|
#include <MG_Util/Async/ShaderCompilePool.h>
|
||||||
|
|
||||||
|
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<double, std::milli>(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<SizeT> shaders;
|
||||||
|
};
|
||||||
|
|
||||||
|
struct Corpus {
|
||||||
|
std::vector<CorpusShader> shaders;
|
||||||
|
std::vector<CorpusProgram> 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<std::string, SizeT> 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<SizeT>(-1);
|
||||||
|
CorpusShader shader;
|
||||||
|
shader.name = name;
|
||||||
|
shader.source = ReadFile(path);
|
||||||
|
shader.stage = StageFromExtension(path.extension().string());
|
||||||
|
if (shader.stage == 0) return static_cast<SizeT>(-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<SizeT>(-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<fs::path> 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<GLuint> 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<GLint>(shader.source.size());
|
||||||
|
GLImpl::ShaderSource(name, 1, &text, &length);
|
||||||
|
shaderNames[i] = name;
|
||||||
|
}
|
||||||
|
|
||||||
|
std::vector<GLuint> 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<Bool> 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<double, std::milli>(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<Uint64>* 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<std::mutex> lock(*m_mutex);
|
||||||
|
m_cv->notify_all();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
std::atomic<Uint64>* 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<Uint64> counter{0};
|
||||||
|
std::mutex mutex;
|
||||||
|
std::condition_variable cv;
|
||||||
|
const auto total = static_cast<Uint64>(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<SharedPtr<JobNode>> nodes;
|
||||||
|
nodes.reserve(jobs);
|
||||||
|
for (SizeT i = 0; i < jobs; ++i) {
|
||||||
|
nodes.push_back(MakeShared<TrivialJob>(&counter, total, &mutex, &cv));
|
||||||
|
}
|
||||||
|
|
||||||
|
const Clock::time_point start = Clock::now();
|
||||||
|
for (auto& node : nodes) pool.Post(Move(node));
|
||||||
|
{
|
||||||
|
std::unique_lock<std::mutex> 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<SizeT>(std::stoull(next()));
|
||||||
|
else if (arg == "--threads") microThreads = static_cast<Uint>(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<unsigned long long>(result.ran),
|
||||||
|
result.ms, result.ms * 1e6 / static_cast<double>(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;
|
||||||
|
}
|
||||||
@@ -10,11 +10,35 @@ target_include_directories(JobNodeTest PRIVATE
|
|||||||
${MGL_ROOT}/MobileGL
|
${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(
|
target_link_libraries(
|
||||||
JobNodeTest PRIVATE
|
JobNodeTest PRIVATE
|
||||||
GTest::gtest_main
|
GTest::gtest
|
||||||
${LINK_LIBRARIES}
|
${LINK_LIBRARIES}
|
||||||
)
|
)
|
||||||
|
|
||||||
include(GoogleTest)
|
include(GoogleTest)
|
||||||
gtest_discover_tests(JobNodeTest DISCOVERY_TIMEOUT 30 PROPERTIES LABELS unit)
|
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}
|
||||||
|
)
|
||||||
|
|||||||
@@ -9,8 +9,19 @@
|
|||||||
#include <gtest/gtest.h>
|
#include <gtest/gtest.h>
|
||||||
|
|
||||||
#include <chrono>
|
#include <chrono>
|
||||||
|
#include <cstdlib>
|
||||||
|
#include <filesystem>
|
||||||
|
#include <fstream>
|
||||||
#include <stdexcept>
|
#include <stdexcept>
|
||||||
|
|
||||||
|
#ifdef _WIN32
|
||||||
|
#include <process.h>
|
||||||
|
#define MGL_TEST_GETPID _getpid
|
||||||
|
#else
|
||||||
|
#include <unistd.h>
|
||||||
|
#define MGL_TEST_GETPID getpid
|
||||||
|
#endif
|
||||||
|
|
||||||
#include "Includes.h"
|
#include "Includes.h"
|
||||||
#include <Config.h>
|
#include <Config.h>
|
||||||
|
|
||||||
@@ -21,6 +32,29 @@ using namespace MobileGL;
|
|||||||
using namespace MobileGL::MG_Util::Async;
|
using namespace MobileGL::MG_Util::Async;
|
||||||
|
|
||||||
namespace {
|
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<char>(file)), std::istreambuf_iterator<char>());
|
||||||
|
}
|
||||||
|
|
||||||
|
std::streamoff LogSize() {
|
||||||
|
std::error_code error;
|
||||||
|
const auto size = std::filesystem::file_size(g_logFilePath, error);
|
||||||
|
return error ? 0 : static_cast<std::streamoff>(size);
|
||||||
|
}
|
||||||
|
|
||||||
// Every test drives its own pool instance rather than ShaderCompilePool::Get(): the
|
// Every test drives its own pool instance rather than ShaderCompilePool::Get(): the
|
||||||
// process-wide pool is stopped permanently by StopAndDrain (that is the teardown
|
// 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.
|
// contract), so a test that drained the singleton would poison every test after it.
|
||||||
@@ -73,6 +107,21 @@ namespace {
|
|||||||
Bool m_open = false;
|
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<SizeT>(
|
||||||
|
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<Bool()>& predicate,
|
Bool WaitUntil(const std::function<Bool()>& predicate,
|
||||||
const std::chrono::milliseconds timeout = std::chrono::seconds(10)) {
|
const std::chrono::milliseconds timeout = std::chrono::seconds(10)) {
|
||||||
const auto deadline = std::chrono::steady_clock::now() + timeout;
|
const auto deadline = std::chrono::steady_clock::now() + timeout;
|
||||||
@@ -89,11 +138,33 @@ namespace {
|
|||||||
// ---------------------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------------------
|
||||||
|
|
||||||
TEST(ShaderCompilePoolLifecycle, ConstructingAPoolStartsNoThreadUntilSomethingIsPosted) {
|
TEST(ShaderCompilePoolLifecycle, ConstructingAPoolStartsNoThreadUntilSomethingIsPosted) {
|
||||||
|
const SizeT before = LiveThreadCount();
|
||||||
|
|
||||||
ShaderCompilePool pool(kTestThreads);
|
ShaderCompilePool pool(kTestThreads);
|
||||||
EXPECT_EQ(pool.GetThreadCount(), kTestThreads);
|
EXPECT_EQ(pool.GetThreadCount(), kTestThreads);
|
||||||
EXPECT_EQ(pool.GetMaxConcurrency(), 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<TestJob>();
|
||||||
|
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) {
|
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<Bool> innerSawPoolThread{false};
|
||||||
|
auto inner = MakeShared<TestJob>(
|
||||||
|
[&](TestJob&) { innerSawPoolThread.store(ShaderCompilePool::IsPoolThread(), std::memory_order_release); });
|
||||||
|
|
||||||
|
std::atomic<Bool> postedFromPoolThread{false};
|
||||||
|
auto outer = MakeShared<TestJob>([&](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<Uint> live{0};
|
||||||
|
std::atomic<Uint> peak{0};
|
||||||
|
std::atomic<Uint> finished{0};
|
||||||
|
|
||||||
|
Vector<SharedPtr<TestJob>> burst;
|
||||||
|
burst.reserve(kBurst);
|
||||||
|
for (Uint i = 0; i < kBurst; ++i) {
|
||||||
|
burst.push_back(MakeShared<TestJob>([&](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<Bool> postedFromPoolThread{false};
|
||||||
|
auto seeder = MakeShared<TestJob>([&](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) {
|
TEST(JobNodeSubmit, ConcurrencyBudgetIsNeverExceeded) {
|
||||||
constexpr Uint kBudget = 2;
|
constexpr Uint kBudget = 2;
|
||||||
constexpr Uint kJobs = 64;
|
constexpr Uint kJobs = 64;
|
||||||
@@ -472,30 +631,58 @@ TEST(JobNodeException, AThrowingJobDoesNotPoisonTheWorkerForLaterJobs) {
|
|||||||
// ---------------------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------------------
|
||||||
|
|
||||||
TEST(ShaderCompilePoolDrain, StopAndDrainWithAThousandQueuedJobsLeavesNoneRunningOrPending) {
|
TEST(ShaderCompilePoolDrain, StopAndDrainWithAThousandQueuedJobsLeavesNoneRunningOrPending) {
|
||||||
constexpr Uint kJobs = 1000;
|
constexpr Uint kQueued = 1000;
|
||||||
ShaderCompilePool pool(kTestThreads);
|
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<SharedPtr<TestJob>> jobs;
|
// Pin that slot with a job that will not return until this test says so. Everything
|
||||||
jobs.reserve(kJobs);
|
// posted behind it is then PROVABLY still in the queue, which is what makes the counts
|
||||||
for (Uint i = 0; i < kJobs; ++i) {
|
// below exact.
|
||||||
jobs.push_back(MakeShared<TestJob>());
|
//
|
||||||
pool.Post(jobs.back());
|
// 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<Bool> entered{false};
|
||||||
|
auto blocker = MakeShared<TestJob>([&](TestJob&) {
|
||||||
|
entered.store(true, std::memory_order_release);
|
||||||
|
gate.Wait();
|
||||||
|
});
|
||||||
|
pool.Post(blocker);
|
||||||
|
ASSERT_TRUE(WaitUntil([&] { return entered.load(); }));
|
||||||
|
|
||||||
|
Vector<SharedPtr<TestJob>> queued;
|
||||||
|
queued.reserve(kQueued);
|
||||||
|
for (Uint i = 0; i < kQueued; ++i) {
|
||||||
|
queued.push_back(MakeShared<TestJob>());
|
||||||
|
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.
|
// The job that was already running still finished: an in-flight body is waited for, not
|
||||||
Uint complete = 0;
|
// interrupted.
|
||||||
Uint cancelled = 0;
|
EXPECT_TRUE(blocker->IsComplete());
|
||||||
for (const auto& job : jobs) {
|
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());
|
ASSERT_TRUE(job->IsTerminal());
|
||||||
if (job->IsComplete()) ++complete;
|
EXPECT_TRUE(job->IsCancelled());
|
||||||
if (job->IsCancelled()) ++cancelled;
|
EXPECT_EQ(job->ran.load(), 0u);
|
||||||
EXPECT_LE(job->ran.load(), 1u);
|
|
||||||
}
|
}
|
||||||
EXPECT_EQ(complete + cancelled, kJobs);
|
|
||||||
EXPECT_GT(cancelled, 0u); // the drain really did abandon queued work rather than run it
|
|
||||||
}
|
}
|
||||||
|
|
||||||
TEST(ShaderCompilePoolDrain, StopAndDrainWaitsForARunningBodyToReturn) {
|
TEST(ShaderCompilePoolDrain, StopAndDrainWaitsForARunningBodyToReturn) {
|
||||||
@@ -535,3 +722,141 @@ TEST(ShaderCompilePoolDrain, JobsPostedAfterADrainStillRun) {
|
|||||||
EXPECT_TRUE(job->IsComplete());
|
EXPECT_TRUE(job->IsComplete());
|
||||||
EXPECT_EQ(job->ran.load(), 1u);
|
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<Bool> sawPoolThread{false};
|
||||||
|
std::thread::id bodyThread{};
|
||||||
|
auto job = MakeShared<TestJob>([&](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<long long>(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;
|
||||||
|
}
|
||||||
|
|||||||
@@ -12,8 +12,14 @@
|
|||||||
#include <asio/post.hpp>
|
#include <asio/post.hpp>
|
||||||
#include <asio/thread_pool.hpp>
|
#include <asio/thread_pool.hpp>
|
||||||
|
|
||||||
|
#include <libfork/core.hpp>
|
||||||
|
#include <libfork/schedule/lazy_pool.hpp>
|
||||||
|
|
||||||
#include <cstdio>
|
#include <cstdio>
|
||||||
|
#include <cstdlib>
|
||||||
#include <deque>
|
#include <deque>
|
||||||
|
#include <functional>
|
||||||
|
#include <span>
|
||||||
|
|
||||||
namespace MobileGL::MG_Util::Async {
|
namespace MobileGL::MG_Util::Async {
|
||||||
namespace {
|
namespace {
|
||||||
@@ -135,46 +141,506 @@ namespace MobileGL::MG_Util::Async {
|
|||||||
return std::clamp(DetectBigCoreCount(), 1u, kMaxAutoShaderCompileThreads);
|
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<char>(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<void()> 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<void()> 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<void()> 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<lf::worker_context*> contexts;
|
||||||
|
std::atomic<Uint64>* cursor;
|
||||||
|
|
||||||
|
void schedule(const lf::submit_handle job) const {
|
||||||
|
const Uint64 index = cursor->fetch_add(1, std::memory_order_relaxed);
|
||||||
|
contexts[static_cast<SizeT>(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<void> {
|
||||||
|
RunLibforkChain(job);
|
||||||
|
co_return;
|
||||||
|
};
|
||||||
|
|
||||||
|
class LibforkJobExecutor final : public JobExecutor {
|
||||||
|
public:
|
||||||
|
explicit LibforkJobExecutor(const Uint threads)
|
||||||
|
: m_pool(static_cast<std::size_t>(std::max(1u, threads))), m_contexts(m_pool.contexts()),
|
||||||
|
m_fallback([this] { FallbackLoop(); }) {}
|
||||||
|
|
||||||
|
~LibforkJobExecutor() override {
|
||||||
|
JoinAll();
|
||||||
|
{
|
||||||
|
const std::lock_guard<std::mutex> 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<void()> fn) override {
|
||||||
|
if (tl_chainOwner == this) {
|
||||||
|
const std::lock_guard<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<void()>& body) noexcept {
|
||||||
|
const std::lock_guard<std::mutex> 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<void()>&& 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<LibforkJob> owned(job);
|
||||||
|
fn = Move(owned->body);
|
||||||
|
throw;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
void Retire() noexcept {
|
||||||
|
// Under the lock, for the reason RetireAndTakeNext spells out.
|
||||||
|
const std::lock_guard<std::mutex> 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<void()> fn;
|
||||||
|
{
|
||||||
|
std::unique_lock<std::mutex> 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<void()>&& 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<lf::worker_context*> m_contexts;
|
||||||
|
std::atomic<Uint64> 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<std::function<void()>> 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<std::function<void()>> 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<LibforkJob> job(raw);
|
||||||
|
LibforkJobExecutor* const owner = job->owner;
|
||||||
|
std::function<void()> 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<JobExecutor> MakeJobExecutor(const AsyncPoolEngine engine, const Uint threads) {
|
||||||
|
switch (engine) {
|
||||||
|
case AsyncPoolEngine::Libfork: return MakeUnique<LibforkJobExecutor>(threads);
|
||||||
|
case AsyncPoolEngine::Asio: break;
|
||||||
|
}
|
||||||
|
return MakeUnique<AsioJobExecutor>(threads);
|
||||||
|
}
|
||||||
|
} // namespace
|
||||||
|
|
||||||
struct ShaderCompilePool::Impl {
|
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;
|
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;
|
std::mutex mutex;
|
||||||
// Created on the first dispatched Post, never in the constructor: asio::thread_pool
|
// Created on the first dispatched Post, never in the constructor: both engines spawn
|
||||||
// spawns its threads eagerly, and a build with async off must not pay for threads it
|
// 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.
|
// will never use.
|
||||||
UniquePtr<asio::thread_pool> pool;
|
UniquePtr<JobExecutor> executor;
|
||||||
std::deque<SharedPtr<JobNode>> queue;
|
std::deque<SharedPtr<JobNode>> queue;
|
||||||
Uint inFlight = 0;
|
Uint inFlight = 0;
|
||||||
Uint maxConcurrency;
|
Uint maxConcurrency;
|
||||||
std::atomic<Bool> stopped{false};
|
std::atomic<Bool> stopped{false};
|
||||||
|
|
||||||
// Callers hold `mutex`. Hands as many queued nodes to Asio as the concurrency budget
|
// Callers hold `mutex`. Hands as many queued nodes to the engine as the concurrency
|
||||||
// allows. Posting under the lock is safe and is what keeps `pool` from being moved
|
// budget allows. Submitting under the lock is safe and is what keeps `executor` from
|
||||||
// out by a concurrent StopAndDrain between the decision and the dispatch: asio::post
|
// being moved out by a concurrent StopAndDrain between the decision and the dispatch:
|
||||||
// only enqueues, it never runs the handler on the calling thread, so it cannot
|
// Submit only enqueues, it never runs the callable on the calling thread, so it cannot
|
||||||
// re-enter this mutex.
|
// 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
|
// 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
|
// added ProgramLinkTask::OnDepSettled as a real one), and a continuation is free to
|
||||||
// call ShaderCompilePool::Post() again. Every caller of DispatchLocked holds `mutex`
|
// call ShaderCompilePool::Post() again. Every caller of DispatchLocked holds `mutex`
|
||||||
// (a plain, non-recursive std::mutex) - Cancel()'ing in here would let that
|
// (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
|
// re-entrant Post() deadlock on the very lock this frame already owns. The caller
|
||||||
// drains `toCancel` after releasing the lock.
|
// 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<SharedPtr<JobNode>>& toCancel) {
|
void DispatchLocked(Vector<SharedPtr<JobNode>>& toCancel) {
|
||||||
while (!queue.empty() && inFlight < maxConcurrency && !stopped.load(std::memory_order_acquire)) {
|
while (!queue.empty() && inFlight < maxConcurrency && !stopped.load(std::memory_order_acquire)) {
|
||||||
// Copy rather than move into the handler: if asio::post throws (it allocates)
|
// Copy rather than move into the callable: if Submit throws (both engines
|
||||||
// the local SharedPtr is still valid, so the node can be settled instead of
|
// allocate) the local SharedPtr is still valid, so the node can be settled
|
||||||
// being stranded Pending in a queue nothing will dispatch from again - a
|
// instead of being stranded Pending in a queue nothing will dispatch from
|
||||||
// joiner would block on it forever. Reclaiming the slot matters just as much:
|
// again - a joiner would block on it forever. Reclaiming the slot matters just
|
||||||
// a leaked `inFlight` shrinks the pool's concurrency budget permanently.
|
// as much: a leaked `inFlight` shrinks the pool's concurrency budget
|
||||||
|
// permanently.
|
||||||
SharedPtr<JobNode> node = queue.front();
|
SharedPtr<JobNode> node = queue.front();
|
||||||
queue.pop_front();
|
queue.pop_front();
|
||||||
++inFlight;
|
++inFlight;
|
||||||
try {
|
try {
|
||||||
asio::post(*pool, [this, node]() mutable { RunOnWorker(Move(node)); });
|
executor->Submit([this, node]() mutable { RunOnWorker(Move(node)); });
|
||||||
} catch (...) {
|
} catch (...) {
|
||||||
--inFlight;
|
--inFlight;
|
||||||
toCancel.push_back(Move(node));
|
toCancel.push_back(Move(node));
|
||||||
@@ -184,7 +650,7 @@ namespace MobileGL::MG_Util::Async {
|
|||||||
|
|
||||||
void RunOnWorker(SharedPtr<JobNode> node) {
|
void RunOnWorker(SharedPtr<JobNode> node) {
|
||||||
tl_isPoolThread = true;
|
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
|
// 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
|
// 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
|
// 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;
|
return m_impl->maxConcurrency;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
AsyncPoolEngine ShaderCompilePool::GetEngine() const { return m_impl->engine; }
|
||||||
|
|
||||||
void ShaderCompilePool::SetMaxConcurrency(const Uint n) {
|
void ShaderCompilePool::SetMaxConcurrency(const Uint n) {
|
||||||
Vector<SharedPtr<JobNode>> toCancel;
|
Vector<SharedPtr<JobNode>> toCancel;
|
||||||
{
|
{
|
||||||
const std::lock_guard<std::mutex> lock(m_impl->mutex);
|
const std::lock_guard<std::mutex> lock(m_impl->mutex);
|
||||||
m_impl->maxConcurrency = std::clamp(n, 1u, m_impl->threadCount);
|
m_impl->maxConcurrency = std::clamp(n, 1u, m_impl->threadCount);
|
||||||
// Raising the budget releases whatever the old one was holding back.
|
// 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.
|
// Outside the lock: see DispatchLocked's comment.
|
||||||
for (const auto& n2 : toCancel) {
|
for (const auto& n2 : toCancel) {
|
||||||
@@ -252,23 +720,23 @@ namespace MobileGL::MG_Util::Async {
|
|||||||
if (!node) return;
|
if (!node) return;
|
||||||
EnsureProcessTeardownSentinel();
|
EnsureProcessTeardownSentinel();
|
||||||
|
|
||||||
// Enqueueing can throw: the thread_pool construction and asio::post both allocate,
|
// Enqueueing can throw: building the engine and submitting to it both allocate (and
|
||||||
// and under memory pressure a throw here would escape glCompileShader leaving the
|
// both spawn threads), and under memory pressure a throw here would escape
|
||||||
// node Pending with nothing left to dispatch it - the first observable read would
|
// glCompileShader leaving the node Pending with nothing left to dispatch it - the
|
||||||
// then block the GL thread forever. Settle the node instead: a cancelled node is a
|
// first observable read would then block the GL thread forever. Settle the node
|
||||||
// state every joiner already handles.
|
// 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
|
// `node` is still valid in the catch for every throw this try can produce. The engine
|
||||||
// thread_pool construction runs before the move; deque::push_back is strongly
|
// construction runs before the move; deque::push_back is strongly exception-safe and
|
||||||
// exception-safe and SharedPtr's move constructor is noexcept, so a throwing
|
// SharedPtr's move constructor is noexcept, so a throwing push_back never consumed it;
|
||||||
// push_back never consumed it; and DispatchLocked contains its own asio::post
|
// and DispatchLocked contains its own Submit failures rather than propagating them
|
||||||
// failures rather than propagating them (see above). Keep it that way.
|
// (see above). Keep it that way.
|
||||||
Bool enqueued = false;
|
Bool enqueued = false;
|
||||||
Vector<SharedPtr<JobNode>> toCancel;
|
Vector<SharedPtr<JobNode>> toCancel;
|
||||||
try {
|
try {
|
||||||
const std::lock_guard<std::mutex> lock(m_impl->mutex);
|
const std::lock_guard<std::mutex> lock(m_impl->mutex);
|
||||||
if (!m_impl->stopped.load(std::memory_order_acquire) && !InProcessTeardown()) {
|
if (!m_impl->stopped.load(std::memory_order_acquire) && !InProcessTeardown()) {
|
||||||
if (!m_impl->pool) m_impl->pool = MakeUnique<asio::thread_pool>(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->queue.push_back(Move(node));
|
||||||
m_impl->DispatchLocked(toCancel);
|
m_impl->DispatchLocked(toCancel);
|
||||||
enqueued = true;
|
enqueued = true;
|
||||||
@@ -304,17 +772,17 @@ namespace MobileGL::MG_Util::Async {
|
|||||||
}
|
}
|
||||||
|
|
||||||
void ShaderCompilePool::StopAndDrain() {
|
void ShaderCompilePool::StopAndDrain() {
|
||||||
// asio::thread_pool::join() from a pool thread would deadlock on itself, and the
|
// Waiting for the workers from a worker would deadlock on itself (asio's join() says
|
||||||
// whole point of this call is that the GL thread waits for the workers.
|
// 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");
|
MOBILEGL_ASSERT(!IsPoolThread(), "ShaderCompilePool::StopAndDrain() called from a pool thread");
|
||||||
|
|
||||||
std::deque<SharedPtr<JobNode>> abandoned;
|
std::deque<SharedPtr<JobNode>> abandoned;
|
||||||
UniquePtr<asio::thread_pool> pool;
|
UniquePtr<JobExecutor> executor;
|
||||||
{
|
{
|
||||||
const std::lock_guard<std::mutex> lock(m_impl->mutex);
|
const std::lock_guard<std::mutex> lock(m_impl->mutex);
|
||||||
m_impl->stopped.store(true, std::memory_order_release);
|
m_impl->stopped.store(true, std::memory_order_release);
|
||||||
abandoned.swap(m_impl->queue);
|
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
|
// 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 (node) node->Cancel();
|
||||||
}
|
}
|
||||||
|
|
||||||
if (pool) {
|
if (executor) {
|
||||||
pool->join(); // returns once every handler already handed to Asio has finished
|
executor->JoinAll(); // returns once every job already handed to the engine is done
|
||||||
pool.reset();
|
executor.reset(); // and this stops the engine's threads
|
||||||
}
|
}
|
||||||
|
|
||||||
const std::lock_guard<std::mutex> lock(m_impl->mutex);
|
const std::lock_guard<std::mutex> lock(m_impl->mutex);
|
||||||
|
|||||||
@@ -11,11 +11,12 @@
|
|||||||
#include <MG_Util/Types.h>
|
#include <MG_Util/Types.h>
|
||||||
#include <MG_Util/Async/JobNode.h>
|
#include <MG_Util/Async/JobNode.h>
|
||||||
|
|
||||||
// This header deliberately includes NO Asio header: asio::thread_pool lives behind the pimpl
|
// This header deliberately includes NO Asio and NO libfork header: both execution engines
|
||||||
// in ShaderCompilePool.cpp. Asio stays a private implementation detail of one translation
|
// live behind the pimpl in ShaderCompilePool.cpp. They stay private implementation details of
|
||||||
// unit, so no consumer target (MG_Test, MG_IntegrationTest, MG_Benchmark - each with its own
|
// one translation unit, so no consumer target (MG_Test, MG_IntegrationTest, MG_Benchmark -
|
||||||
// target_include_directories) needs the Asio include path, and no consumer pays its compile
|
// each with its own target_include_directories) needs either include path, and no consumer
|
||||||
// time. Do not add one here.
|
// 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 {
|
namespace MobileGL::MG_Util::Async {
|
||||||
// Stage 7: on by default. The gate behind the flip (2026-08-09, headless Mesa, both
|
// 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.
|
// MOBILEGL_ASYNC_SHADER_COMPILE_THREADS overrides it outright.
|
||||||
Uint DetectShaderCompileThreadCount();
|
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 {
|
class ShaderCompilePool {
|
||||||
public:
|
public:
|
||||||
explicit ShaderCompilePool(Uint threadCount);
|
explicit ShaderCompilePool(Uint threadCount);
|
||||||
@@ -97,6 +123,11 @@ namespace MobileGL::MG_Util::Async {
|
|||||||
Uint GetThreadCount() const;
|
Uint GetThreadCount() const;
|
||||||
Uint GetMaxConcurrency() 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
|
// Bounded concurrency doubles as the memory bound, and is how
|
||||||
// glMaxShaderCompilerThreadsKHR(n) is honoured: a 300-program pack load cannot put
|
// glMaxShaderCompilerThreadsKHR(n) is honoured: a 300-program pack load cannot put
|
||||||
// 300 glslang arenas in flight at once. Clamped to [1, thread count].
|
// 300 glslang arenas in flight at once. Clamped to [1, thread count].
|
||||||
|
|||||||
@@ -151,12 +151,20 @@ namespace MobileGL::MG_Util::SelfTest {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const Uint threads = MG_Util::Async::DetectShaderCompileThreadCount();
|
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,
|
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 "
|
"and GL_MAX_SHADER_COMPILER_THREADS_KHR = {} (set environment variable "
|
||||||
"MOBILEGL_ASYNC_SHADER_COMPILE=0 to disable it, or "
|
"MOBILEGL_ASYNC_SHADER_COMPILE=0 to disable it, "
|
||||||
"MOBILEGL_ASYNC_SHADER_COMPILE_THREADS=n to change the count)",
|
"MOBILEGL_ASYNC_SHADER_COMPILE_THREADS=n to change the count, or "
|
||||||
threads, threads == 1 ? "" : "s", threads));
|
"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.
|
// Appends the four "MobileGL reported ..." rows for one backend section.
|
||||||
|
|||||||
Reference in New Issue
Block a user