"""PipelineManager: owns the forward renderer's 3D graphics pipelines.
Extracted from Renderer so pipeline creation, rebuild-on-HDR-switch,
resize, and cleanup live in one place. Renderer re-exports the
pipeline handles as attributes so SceneContentRenderer keeps accessing
``renderer._pipeline`` etc. without change.
"""
from typing import Any
import vulkan as vk
from ..gpu.pipeline import (
MESH_PUSH_CONSTANT_SIZE,
PipelineSpec,
build_pipeline,
create_shader_module,
)
from ..materials.shader_compiler import compile_shader
from .vertex_layouts import EXTRAS_BINDING, MESH_BINDINGS, SKINNED_MESH_BINDINGS
# Forward mesh pipelines with the extras stream bound (tangent normal mapping):
# streams 0+1 plus the tangent/colour/uv2 stream at binding 2.
TANGENT_MESH_BINDINGS = (*MESH_BINDINGS, EXTRAS_BINDING)
__all__ = ["PipelineManager"]
[docs]
class PipelineManager:
"""Owns the 3D forward-render pipelines and their shader modules."""
def __init__(self, engine: Any) -> None:
self._engine = engine
self.vert_module: Any = None
self.frag_module: Any = None
# Thin G-buffer: a second frag permutation of cube_textured.frag
# compiled with -DGBUFFER (emits octahedral normal + roughness into
# attachment 1). Compiled lazily the first time the G-buffer is active so
# a renderer that never enables a consumer pays nothing.
self.gbuffer_frag_module: Any = None
self.skinned_vert_module: Any = None
# Tangent normal mapping: HAS_TANGENTS permutations of
# cube.vert / cube_textured.frag, compiled lazily the first time a mesh
# with real tangents draws, so tangent-free scenes pay nothing.
self.tangent_vert_module: Any = None
self.tangent_frag_module: Any = None
self.tangent_gbuffer_frag_module: Any = None
self.opaque: Any = None
self.opaque_layout: Any = None
self.nocull: Any = None
self.nocull_layout: Any = None
self.transparent: Any = None
self.transparent_layout: Any = None
self.skinned: Any = None
self.skinned_layout: Any = None
# Tangent variants of the opaque / double-sided / transparent pipelines
# (extras stream at binding 2). None until first requested; rebuilt with
# the base triple on resize / HDR / G-buffer toggles once active.
self.tangent_opaque: Any = None
self.tangent_opaque_layout: Any = None
self.tangent_nocull: Any = None
self.tangent_nocull_layout: Any = None
self.tangent_transparent: Any = None
self.tangent_transparent_layout: Any = None
self._tangent_active = False
self._ssbo_layout: Any = None
self._joint_layout: Any = None
# Render pass + extent the pipelines were last built against, so the
# lazily-built tangent triple always matches the live pass.
self._render_pass: Any = None
self._extent: Any = None
# Reflection-probe blend width: the top-K
# probe blend in cube_textured.frag is a compile-time constant. At the
# default (2) no -D flag is passed, so the compiled artefacts and the
# built pipelines are byte-identical to the pre-knob renderer.
self._probe_blend_count = 2
[docs]
def setup(self, ssbo_layout: Any, joint_layout: Any) -> None:
"""Compile shaders and build the four 3D pipelines against the engine render pass."""
self._ssbo_layout = ssbo_layout
self._joint_layout = joint_layout
e = self._engine
device = e.ctx.device
shader_dir = e.shader_dir
vert_spv = compile_shader(shader_dir / "cube.vert")
frag_spv = compile_shader(shader_dir / "cube_textured.frag", defines=self._frag_defines())
self.vert_module = create_shader_module(device, vert_spv)
self.frag_module = create_shader_module(device, frag_spv)
skinned_vert_spv = compile_shader(shader_dir / "skinned.vert")
self.skinned_vert_module = create_shader_module(device, skinned_vert_spv)
self._build_pipelines(e.render_pass, e.extent)
def _frag_defines(self, *extra: str) -> tuple[str, ...]:
"""Defines for a cube_textured.frag permutation, folding in the probe
blend width. The default (2) adds no flag: the base permutations stay
byte-identical to the pre-knob shader cache."""
if self._probe_blend_count != 2:
return (*extra, f"PROBE_BLEND_COUNT={self._probe_blend_count}")
return extra
[docs]
def set_probe_blend_count(self, count: int) -> None:
"""Set the top-K reflection-probe blend width.
``PROBE_BLEND_COUNT`` is a compile-time constant in cube_textured.frag,
so a change recompiles the lit fragment permutations and rebuilds the
mesh pipelines (device-wait-idle; rare transition, driven by
``WorldEnvironment.probe_blend_count``). No-op when unchanged. Values
are clamped to [1, 8] (0 = probes-off is handled by the capture gate,
never by a shader permutation: a zero-length top-K array is invalid
GLSL, and with zero live probes the blend loop is already a no-op).
"""
count = max(1, min(int(count), 8))
if count == self._probe_blend_count:
return
self._probe_blend_count = count
if self.frag_module is None:
return # before setup(): the first build picks the value up
e = self._engine
device = e.ctx.device
vk.vkDeviceWaitIdle(device)
for mod in (
self.frag_module, self.gbuffer_frag_module,
self.tangent_frag_module, self.tangent_gbuffer_frag_module,
):
if mod:
vk.vkDestroyShaderModule(device, mod, None)
self.frag_module = self.gbuffer_frag_module = None
self.tangent_frag_module = self.tangent_gbuffer_frag_module = None
frag_spv = compile_shader(e.shader_dir / "cube_textured.frag", defines=self._frag_defines())
self.frag_module = create_shader_module(device, frag_spv)
# Rebuild against the live pass/extent; the G-buffer + tangent
# permutations recompile lazily inside the build when active.
self._destroy_pipelines()
self._build_pipelines(self._render_pass, self._extent)
[docs]
def rebuild_for_render_pass(self, render_pass: Any) -> None:
"""Rebuild all four pipelines against ``render_pass`` (e.g. HDR target)."""
self._destroy_pipelines()
self._build_pipelines(render_pass, self._engine.extent)
[docs]
def rebuild_for_resize(self, width: int, height: int, render_pass: Any) -> None:
"""Rebuild all four pipelines for a new framebuffer size."""
self._destroy_pipelines()
self._build_pipelines(render_pass, (width, height))
[docs]
def cleanup(self) -> None:
"""Destroy pipelines, layouts, and shader modules."""
self._destroy_pipelines()
device = self._engine.ctx.device
for mod in (
self.skinned_vert_module, self.vert_module, self.frag_module, self.gbuffer_frag_module,
self.tangent_vert_module, self.tangent_frag_module, self.tangent_gbuffer_frag_module,
):
if mod:
vk.vkDestroyShaderModule(device, mod, None)
self.skinned_vert_module = self.vert_module = self.frag_module = None
self.gbuffer_frag_module = None
self.tangent_vert_module = self.tangent_frag_module = self.tangent_gbuffer_frag_module = None
def _build_pipelines(self, render_pass: Any, extent: Any) -> None:
e = self._engine
device = e.ctx.device
tex_layout = e.texture_descriptor_layout
self._render_pass = render_pass
self._extent = extent
# Thin G-buffer: when a consumer is active the HDR pass carries a
# second colour attachment, so every forward pipeline must declare two
# blend attachments. The lit uber writes the octahedral normal +
# roughness (GBUFFER frag permutation, writes_gbuffer=True); the
# transparent pass shares the two-attachment layout but masks the normal
# target (writes_gbuffer=False). When inactive, n_att=1 and the historical
# single-output frag module builds byte-identical pipelines.
gbuffer = bool(getattr(e, "gbuffer_active", False))
n_att = 2 if gbuffer else 1
if gbuffer and self.gbuffer_frag_module is None:
gbuffer_spv = compile_shader(e.shader_dir / "cube_textured.frag", defines=self._frag_defines("GBUFFER"))
self.gbuffer_frag_module = create_shader_module(device, gbuffer_spv)
lit_frag = self.gbuffer_frag_module if gbuffer else self.frag_module
# Forward mesh pipelines share the split position + shading vertex
# streams (vertex_layouts.MESH_BINDINGS), the mesh push-constant block,
# and the SSBO + bindless-texture descriptor sets; opaque/nocull differ
# only in backface culling, transparent adds alpha blending with
# depth-write disabled.
forward_base = {
"topology": vk.VK_PRIMITIVE_TOPOLOGY_TRIANGLE_LIST,
"vertex_bindings": MESH_BINDINGS,
"set_layouts": (self._ssbo_layout, tex_layout),
"push_size": MESH_PUSH_CONSTANT_SIZE,
"attachment_count": n_att,
}
self.opaque, self.opaque_layout = build_pipeline(
device,
PipelineSpec(
name="Forward", cull_mode=vk.VK_CULL_MODE_BACK_BIT,
writes_gbuffer=gbuffer, **forward_base,
),
render_pass, extent,
vert_module=self.vert_module, frag_module=lit_frag,
)
self.nocull, self.nocull_layout = build_pipeline(
device,
PipelineSpec(
name="Forward (double-sided)", cull_mode=vk.VK_CULL_MODE_NONE,
writes_gbuffer=gbuffer, **forward_base,
),
render_pass, extent,
vert_module=self.vert_module, frag_module=lit_frag,
)
self.transparent, self.transparent_layout = build_pipeline(
device,
PipelineSpec(
name="Transparent",
cull_mode=vk.VK_CULL_MODE_NONE, # transparent objects often need both sides
depth_write=False, # depth test stays on
blend="alpha",
dst_alpha_factor=vk.VK_BLEND_FACTOR_ONE_MINUS_SRC_ALPHA,
# Transparent geometry does not write the normal target (blended
# normals are meaningless); its attachment 1 is masked.
writes_gbuffer=False,
**forward_base,
),
render_pass, extent,
vert_module=self.vert_module, frag_module=lit_frag,
)
self.skinned, self.skinned_layout = build_pipeline(
device,
PipelineSpec(
name="Skinned",
topology=vk.VK_PRIMITIVE_TOPOLOGY_TRIANGLE_LIST,
vertex_bindings=SKINNED_MESH_BINDINGS,
# Sets: 0=SSBOs, 1=textures, 2=joint matrices SSBO.
set_layouts=(self._ssbo_layout, tex_layout, self._joint_layout),
push_size=MESH_PUSH_CONSTANT_SIZE,
cull_mode=vk.VK_CULL_MODE_BACK_BIT,
attachment_count=n_att,
writes_gbuffer=gbuffer,
),
render_pass, extent,
vert_module=self.skinned_vert_module, frag_module=lit_frag,
)
# Tangent triple: rebuilt with the base pipelines once any tangent mesh
# has activated it (resize / HDR / G-buffer toggles land here too).
if self._tangent_active:
self._build_tangent_pipelines()
[docs]
def tangent_pipelines(self) -> tuple[Any, Any, Any]:
"""The (opaque, double-sided, transparent) tangent pipeline triple.
Built lazily on first request (compiling the HAS_TANGENTS shader
permutations), then kept in sync with the base pipelines by every
rebuild. Callers only ask when a registered mesh carries real tangents,
so a tangent-free renderer never compiles or creates any of this.
"""
if not self._tangent_active:
self._tangent_active = True
self._build_tangent_pipelines()
return self.tangent_opaque, self.tangent_nocull, self.tangent_transparent
def _build_tangent_pipelines(self) -> None:
e = self._engine
device = e.ctx.device
gbuffer = bool(getattr(e, "gbuffer_active", False))
if self.tangent_vert_module is None:
vert_spv = compile_shader(e.shader_dir / "cube.vert", defines=("HAS_TANGENTS",))
self.tangent_vert_module = create_shader_module(device, vert_spv)
if gbuffer and self.tangent_gbuffer_frag_module is None:
spv = compile_shader(e.shader_dir / "cube_textured.frag",
defines=self._frag_defines("GBUFFER", "HAS_TANGENTS"))
self.tangent_gbuffer_frag_module = create_shader_module(device, spv)
if not gbuffer and self.tangent_frag_module is None:
spv = compile_shader(e.shader_dir / "cube_textured.frag", defines=self._frag_defines("HAS_TANGENTS"))
self.tangent_frag_module = create_shader_module(device, spv)
lit_frag = self.tangent_gbuffer_frag_module if gbuffer else self.tangent_frag_module
tangent_base = {
"topology": vk.VK_PRIMITIVE_TOPOLOGY_TRIANGLE_LIST,
"vertex_bindings": TANGENT_MESH_BINDINGS,
"set_layouts": (self._ssbo_layout, e.texture_descriptor_layout),
"push_size": MESH_PUSH_CONSTANT_SIZE,
"attachment_count": 2 if gbuffer else 1,
}
self.tangent_opaque, self.tangent_opaque_layout = build_pipeline(
device,
PipelineSpec(
name="Forward (tangent)", cull_mode=vk.VK_CULL_MODE_BACK_BIT,
writes_gbuffer=gbuffer, **tangent_base,
),
self._render_pass, self._extent,
vert_module=self.tangent_vert_module, frag_module=lit_frag,
)
self.tangent_nocull, self.tangent_nocull_layout = build_pipeline(
device,
PipelineSpec(
name="Forward (double-sided, tangent)", cull_mode=vk.VK_CULL_MODE_NONE,
writes_gbuffer=gbuffer, **tangent_base,
),
self._render_pass, self._extent,
vert_module=self.tangent_vert_module, frag_module=lit_frag,
)
self.tangent_transparent, self.tangent_transparent_layout = build_pipeline(
device,
PipelineSpec(
name="Transparent (tangent)",
cull_mode=vk.VK_CULL_MODE_NONE,
depth_write=False,
blend="alpha",
dst_alpha_factor=vk.VK_BLEND_FACTOR_ONE_MINUS_SRC_ALPHA,
writes_gbuffer=False, # transparent masks the normal target (parity with the base triple)
**tangent_base,
),
self._render_pass, self._extent,
vert_module=self.tangent_vert_module, frag_module=lit_frag,
)
def _destroy_pipelines(self) -> None:
device = self._engine.ctx.device
for pipeline, layout in (
(self.opaque, self.opaque_layout),
(self.nocull, self.nocull_layout),
(self.transparent, self.transparent_layout),
(self.skinned, self.skinned_layout),
(self.tangent_opaque, self.tangent_opaque_layout),
(self.tangent_nocull, self.tangent_nocull_layout),
(self.tangent_transparent, self.tangent_transparent_layout),
):
if pipeline:
vk.vkDestroyPipeline(device, pipeline, None)
if layout:
vk.vkDestroyPipelineLayout(device, layout, None)
self.opaque = self.opaque_layout = None
self.nocull = self.nocull_layout = None
self.transparent = self.transparent_layout = None
self.skinned = self.skinned_layout = None
self.tangent_opaque = self.tangent_opaque_layout = None
self.tangent_nocull = self.tangent_nocull_layout = None
self.tangent_transparent = self.tangent_transparent_layout = None