diff --git a/src/graphics/host_gpu/renderer/pipeline/pipelineCache.h b/src/graphics/host_gpu/renderer/pipeline/pipelineCache.h index 49030e173..b9ab8981a 100644 --- a/src/graphics/host_gpu/renderer/pipeline/pipelineCache.h +++ b/src/graphics/host_gpu/renderer/pipeline/pipelineCache.h @@ -5,6 +5,7 @@ #include "common/assert.h" #include "common/common.h" #include "common/file.h" +#include "common/uniqueFunction.h" #include "graphics/host_gpu/renderer/colorRenderTarget.h" #include "graphics/host_gpu/renderer/depthRenderTarget.h" #include "graphics/host_gpu/renderer/renderTarget.h" @@ -14,10 +15,14 @@ #include "graphics/shader/shader.h" #include +#include #include #include #include +#include +#include #include +#include #include #include #include @@ -141,6 +146,29 @@ struct DrawRenderState { } }; +// Writes the shader store on its own thread, running jobs in the order they were posted. +class ShaderStoreWriter { +public: + // Keeps the first `kept` bytes of the store at `path`, or starts it over with `header`. The + // thread runs only if that succeeded. + void Open(const std::filesystem::path& path, std::span header, uint64_t kept); + explicit operator bool() const { return m_thread.joinable(); } + // Compresses `record` and appends it on the store thread, in call order. + void Append(std::vector&& record); + +private: + friend struct PipelineCacheTestAccess; + + void Post(Common::UniqueFunction&& job); + + Common::File m_file; + std::mutex m_mutex; + std::condition_variable_any m_ready; + std::queue> m_jobs; + // Declared last, so it drains the queue and joins before the rest is destroyed. + std::jthread m_thread; +}; + // The owning renderer serializes access, including saves while the GPU is running. class PipelineCache { public: @@ -219,8 +247,8 @@ private: std::unordered_map m_graphics_pipelines; std::unordered_map m_compute_pipelines; std::unordered_map m_tess_layouts; - // Appends records of everything first created; invalid without a shader store. - Common::File m_store; + // Records everything first created; closed without a shader store. + ShaderStoreWriter m_store; template const ShaderModule* GetProgram(const ShaderParams& params, InputInfo& input_info, diff --git a/src/graphics/host_gpu/renderer/pipeline/pipelineSerialization.cpp b/src/graphics/host_gpu/renderer/pipeline/pipelineSerialization.cpp index ec5b94a92..ef33bf154 100644 --- a/src/graphics/host_gpu/renderer/pipeline/pipelineSerialization.cpp +++ b/src/graphics/host_gpu/renderer/pipeline/pipelineSerialization.cpp @@ -72,19 +72,6 @@ std::vector StoreHeader(const GraphicContext& graphics) { return std::move(header.bytes); } -void Append(Common::File& store, const Writer& record) { - std::vector bytes(sizeof(Frame) + ZSTD_compressBound(record.bytes.size())); - const auto size = ZSTD_compress(bytes.data() + sizeof(Frame), bytes.size() - sizeof(Frame), - record.bytes.data(), record.bytes.size(), 3); - EXIT_IF(ZSTD_isError(size)); - const Frame frame {.compressed_size = static_cast(size), - .raw_size = static_cast(record.bytes.size()), - .checksum = XXH3_64bits(bytes.data() + sizeof(Frame), size)}; - std::memcpy(bytes.data(), &frame, sizeof(frame)); - // A failed write leaves a partial record; the next run drops it and everything after it. - store.Write(bytes.data(), static_cast(sizeof(Frame) + size)); -} - // Record fields, shared by writing and reading. template void IoKey(S& io, Key& key) { auto& [stage, hash, user_data_count, static_state] = key; @@ -99,6 +86,56 @@ template void IoModule(S& io, Module& stage) { } // namespace +void ShaderStoreWriter::Open(const std::filesystem::path& path, std::span header, + uint64_t kept) { + Common::File::CreateDirectories(path.parent_path()); + if (kept == 0) { + if (m_file.Create(path)) { + m_file.Write(header.data(), static_cast(header.size())); + } + } else if (m_file.Open(path, Common::File::Mode::ReadWrite)) { + m_file.Truncate(kept); + m_file.Seek(kept); + } + if (m_file.IsInvalid()) { + return; + } + m_thread = std::jthread([this](std::stop_token stop) { + std::unique_lock lock(m_mutex); + // Once stop is requested the wait returns false only when the queue is empty. + while (m_ready.wait(lock, stop, [this] { return !m_jobs.empty(); })) { + const auto job = std::move(m_jobs.front()); + m_jobs.pop(); + lock.unlock(); + job(); + lock.lock(); + } + }); +} + +void ShaderStoreWriter::Append(std::vector&& record) { + Post([this, record = std::move(record)] { + std::vector bytes(sizeof(Frame) + ZSTD_compressBound(record.size())); + const auto size = ZSTD_compress(bytes.data() + sizeof(Frame), bytes.size() - sizeof(Frame), + record.data(), record.size(), 3); + EXIT_IF(ZSTD_isError(size)); + const Frame frame {.compressed_size = static_cast(size), + .raw_size = static_cast(record.size()), + .checksum = XXH3_64bits(bytes.data() + sizeof(Frame), size)}; + std::memcpy(bytes.data(), &frame, sizeof(frame)); + // A failed write leaves a partial record; the next run drops it and everything after it. + m_file.Write(bytes.data(), static_cast(sizeof(Frame) + size)); + }); +} + +void ShaderStoreWriter::Post(Common::UniqueFunction&& job) { + { + std::scoped_lock lock(m_mutex); + m_jobs.push(std::move(job)); + } + m_ready.notify_one(); +} + void PipelineCache::WarmUp(const std::filesystem::path& path) { if (path.empty()) { return; @@ -126,16 +163,7 @@ void PipelineCache::WarmUp(const std::filesystem::path& path) { } } // A damaged record ends the store; the records before it stay. - const auto kept = reusable ? bytes.size() - records.size() : 0; - Common::File::CreateDirectories(path.parent_path()); - if (kept == 0) { - if (m_store.Create(path)) { - m_store.Write(header.data(), static_cast(header.size())); - } - } else if (m_store.Open(path, Common::File::Mode::ReadWrite)) { - m_store.Truncate(kept); - m_store.Seek(kept); - } + m_store.Open(path, header, reusable ? bytes.size() - records.size() : 0); const auto modules = MapModules(); for (const auto& key: pipelines) { @@ -238,9 +266,10 @@ bool PipelineCache::LoadRecord(std::span& records, return true; } +// Records are encoded here, since programs and modules change later. void PipelineCache::StoreModule(const ProgramKey& key, const Program& program, const ShaderModule& module, const std::vector& spirv) { - if (m_store.IsInvalid()) { + if (!m_store) { return; } Writer record; @@ -249,26 +278,26 @@ void PipelineCache::StoreModule(const ProgramKey& key, const Program& program, IoAll(record, ShaderRecompiler::IR::SaveResourcePlan(program.resource_plan)); IoModule(record, module); IoAll(record, spirv); - Append(m_store, record); + m_store.Append(std::move(record.bytes)); } void PipelineCache::StorePipeline(const GraphicsPipelineKey& key) { - if (m_store.IsInvalid()) { + if (!m_store) { return; } Writer record; IoAll(record, Record::GraphicsPipeline); record.Raw(&key, sizeof(key)); - Append(m_store, record); + m_store.Append(std::move(record.bytes)); } void PipelineCache::StoreLayout(uint64_t key, const ShaderTessellationInputInfo& layout) { - if (m_store.IsInvalid()) { + if (!m_store) { return; } Writer record; IoAll(record, Record::TessellationLayout, key, layout); - Append(m_store, record); + m_store.Append(std::move(record.bytes)); } PipelineCache::ModuleMap PipelineCache::MapModules() const { diff --git a/tests/ShaderRecompilerComputeTests.cpp b/tests/ShaderRecompilerComputeTests.cpp index 98a184762..ce4f6eb1b 100644 --- a/tests/ShaderRecompilerComputeTests.cpp +++ b/tests/ShaderRecompilerComputeTests.cpp @@ -611,6 +611,13 @@ struct PipelineCacheTestAccess { cache.StoreLayout(key, layout); } + // Holds the store thread until the cache is destroyed. + static void HoldStoreUntilExit(PipelineCache &cache) { + cache.m_store.Post([stop = cache.m_store.m_thread.get_stop_token()] { + while (!stop.stop_requested()) std::this_thread::yield(); + }); + } + static const auto &TessLayouts(const PipelineCache &cache) { return cache.m_tess_layouts; } static std::array Pipelines(const PipelineCache &cache) { @@ -44917,11 +44924,20 @@ void CheckShaderStore(VulkanHarness &vulkan) { flip(0); Require(name, "stale store", run() == cold && std::filesystem::file_size(store) == size, "a store with another header was reused instead of started over"); - // Reflecting a layout decodes LS and HS, so a restart takes it from the store. + // Reflecting a layout decodes LS and HS, so a restart takes it from the store. Records land + // while the cache lives; one still queued at its destruction lands before it is gone. const ShaderTessellationInputInfo layout{1, 2, 3, 4, 5, 6, 7, 8, 9}; { RenderContext context(vulkan.RuntimeContext(), store); - PipelineCacheTestAccess::StoreLayout(context.GetPipelineCache(), 1, layout); + auto &cache = context.GetPipelineCache(); + PipelineCacheTestAccess::StoreLayout(cache, 1, layout); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (std::filesystem::file_size(store) <= size && std::chrono::steady_clock::now() < deadline) + std::this_thread::yield(); + Require(name, "background write", std::filesystem::file_size(store) > size, + "a record waited for the cache to be destroyed"); + PipelineCacheTestAccess::HoldStoreUntilExit(cache); + PipelineCacheTestAccess::StoreLayout(cache, 2, layout); } { RenderContext context(vulkan.RuntimeContext(), store); @@ -44929,6 +44945,8 @@ void CheckShaderStore(VulkanHarness &vulkan) { Require(name, "tessellation layout", layouts.contains(1) && std::memcmp(&layouts.at(1), &layout, sizeof(layout)) == 0, "a restart did not load the stored tessellation layout"); + Require(name, "drain at exit", layouts.contains(2), + "a record still queued when the cache was destroyed was dropped"); } std::filesystem::remove_all(store.parent_path()); Require(name, "release",