pipeline: create shader-store pipelines at boot on half the hardware threads

src/common/parallelFor.h

@@ -0,0 +1,35 @@
+#ifndef KYTY_COMMON_PARALLELFOR_H_
+#define KYTY_COMMON_PARALLELFOR_H_
+
+#include <algorithm>
+#include <atomic>
+#include <cstddef>
+#include <thread>
+#include <vector>
+
+namespace Common {
+
+// Half the hardware threads, at least one.
+inline size_t ParallelThreads() {
+	return std::max(1u, std::thread::hardware_concurrency() / 2);
+}
+
+// Calls job(i) for every i below count on up to ParallelThreads() threads, the caller included.
+template <typename Job>
+void ParallelFor(size_t count, const Job& job) {
+	std::atomic_size_t next = 0;
+	const auto         work = [&] {
+		for (auto i = next++; i < count; i = next++) {
+			job(i);
+		}
+	};
+	std::vector<std::jthread> helpers;
+	for (auto i = std::min(count, ParallelThreads()); i > 1; i--) {
+		helpers.emplace_back(work);
+	}
+	work();
+}
+
+} // namespace Common
+
+#endif // KYTY_COMMON_PARALLELFOR_H_

src/graphics/host_gpu/renderer/pipeline/pipelineCache.h

@@ -262,7 +262,7 @@ private:
 
 	// pipelineSerialization.cpp
 	void WarmUp(const std::filesystem::path& path);
-	bool LoadRecord(std::span<const uint8_t>& records, std::vector<GraphicsPipelineKey>& pipelines);
+	bool LoadRecord(std::span<const uint8_t>& records);
 	void StoreModule(const ProgramKey& key, const Program& program, const ShaderModule& module,
 	                 const std::vector<uint32_t>& spirv);
 	void StorePipeline(const GraphicsPipelineKey& key);

src/graphics/host_gpu/renderer/pipeline/pipelineSerialization.cpp

@@ -1,6 +1,7 @@
 #include "graphics/host_gpu/renderer/pipeline/pipelineCache.h"
 
 #include "common/logging/log.h"
+#include "common/parallelFor.h"
 #include "common/stringUtils.h"
 #include "graphics/host_gpu/vulkanCommon.h"
 #include "graphics/shader/recompiler/ShaderRecompiler.h"
@@ -153,35 +154,41 @@ void PipelineCache::WarmUp(const std::filesystem::path& path) {
 		file.Read(bytes.data(), static_cast<uint32_t>(bytes.size()), &read);
 		bytes.resize(read);
 	}
-	Log::WriteToConsoleAndLog(
+	Log::WriteToConsoleAndLog(fmt::format(
-	    fmt::format("Shader cache: compiling shaders and pipelines from {} ({:.1f} MB)\n",
+	    "Shader cache: compiling shaders and pipelines from {} ({:.1f} MB) with {} threads\n",
-	                Common::PathToString(path), static_cast<double>(bytes.size()) / 1'000'000.0));
+	    Common::PathToString(path), static_cast<double>(bytes.size()) / 1'000'000.0,
-	std::vector<GraphicsPipelineKey> pipelines;
+	    Common::ParallelThreads()));
-	auto                             records = std::span<const uint8_t>(bytes);
+	auto       records = std::span<const uint8_t>(bytes);
 	const bool reusable =
 	    records.size() >= header.size() && std::ranges::equal(records.first(header.size()), header);
 	if (reusable) {
 		records = records.subspan(header.size());
-		while (LoadRecord(records, pipelines)) {
+		while (LoadRecord(records)) {
 		}
 	}
 	// A damaged record ends the store; the records before it stay.
 	m_store.Open(path, header, reusable ? bytes.size() - records.size() : 0);
 
 	const auto modules = MapModules();
-	for (const auto& key: pipelines) {
+	std::vector<Common::UniqueFunction<void>> jobs;
-		Pipeline pipeline;
+	jobs.reserve(m_graphics_pipelines.size() + modules.size());
-		if (CreateGraphicsPipeline(pipeline, key, modules)) {
+	for (auto& node: m_graphics_pipelines) {
-			m_graphics_pipelines.emplace(key, pipeline);
+		jobs.emplace_back([this, &modules, &node] {
-		}
+			CreateGraphicsPipeline(node.second, node.first, modules);
+		});
 	}
 	// A compute pipeline depends only on its module.
 	for (const auto* module: modules | std::views::values) {
 		if (module != nullptr && module->info.stage == ShaderType::Compute) {
-			CreatePipelineInternal(m_graphics, m_compute_pipelines[module->id], *module,
+			jobs.emplace_back([this, module, &pipeline = m_compute_pipelines[module->id]] {
-			                       m_driver_cache);
+				CreatePipelineInternal(m_graphics, pipeline, *module, m_driver_cache);
+			});
 		}
 	}
+	Common::ParallelFor(jobs.size(), [&](size_t i) { jobs[i](); });
+	// A key with a stage missing from the store keeps no pipeline.
+	std::erase_if(m_graphics_pipelines,
+	              [](const auto& entry) { return entry.second.pipeline == nullptr; });
 	if (!m_graphics_pipelines.empty() || !m_compute_pipelines.empty()) {
 		CheckpointDriverCache();
 	}
@@ -194,8 +201,7 @@ void PipelineCache::WarmUp(const std::filesystem::path& path) {
 	        .count()));
 }
 
-bool PipelineCache::LoadRecord(std::span<const uint8_t>&        records,
+bool PipelineCache::LoadRecord(std::span<const uint8_t>& records) {
-                               std::vector<GraphicsPipelineKey>& pipelines) {
 	Frame frame;
 	if (records.size() < sizeof(frame)) {
 		return false;
@@ -249,7 +255,8 @@ bool PipelineCache::LoadRecord(std::span<const uint8_t>&        records,
 			if (!complete()) {
 				return false;
 			}
-			pipelines.push_back(key);
+			// WarmUp creates the pipeline once every record is loaded.
+			m_graphics_pipelines.try_emplace(key);
 			break;
 		}
 		case Record::TessellationLayout: {