Source code for simvx.graphics.gpu.sync

"""Fences, semaphores, and frames-in-flight synchronization."""


from __future__ import annotations

import logging
from typing import Any

import vulkan as vk

from .._types import FRAMES_IN_FLIGHT

__all__ = ["FrameSync"]

log = logging.getLogger(__name__)




class FrameSync:
    """Per-frame synchronization primitives.

    Uses FRAMES_IN_FLIGHT fences and acquire semaphores, plus one
    render-finished semaphore per swapchain image to avoid the
    semaphore-reuse hazard described in the Vulkan swapchain guide.
    """

    def __init__(self, device: Any, image_count: int, frame_count: int = FRAMES_IN_FLIGHT) -> None:
        self.device = device
        self.frame_count = frame_count
        self.image_count = image_count
        self.current_frame = 0

        # Per frame-in-flight
        self.fences: list[Any] = []
        self.image_available: list[Any] = []

        # Per swapchain image
        self.render_finished: list[Any] = []

        # Track which fence guards each swapchain image
        self.images_in_flight: list[Any] = []



    def create(self) -> None:
        fence_info = vk.VkFenceCreateInfo(flags=vk.VK_FENCE_CREATE_SIGNALED_BIT)
        sem_info = vk.VkSemaphoreCreateInfo()

        for _ in range(self.frame_count):
            self.fences.append(vk.vkCreateFence(self.device, fence_info, None))
            self.image_available.append(vk.vkCreateSemaphore(self.device, sem_info, None))

        for _ in range(self.image_count):
            self.render_finished.append(vk.vkCreateSemaphore(self.device, sem_info, None))

        self.images_in_flight = [None] * self.image_count
        log.debug("Frame sync created (%d frames, %d images)", self.frame_count, self.image_count)




    def wait_and_reset(self) -> None:
        fence = self.fences[self.current_frame]
        vk.vkWaitForFences(self.device, 1, [fence], vk.VK_TRUE, 2_000_000_000)
        vk.vkResetFences(self.device, 1, [fence])




    def wait_for_image(self, image_index: int) -> None:
        """Wait for any in-flight work using this swapchain image."""
        fence = self.images_in_flight[image_index]
        if fence is not None and fence != self.fences[self.current_frame]:
            vk.vkWaitForFences(self.device, 1, [fence], vk.VK_TRUE, 2_000_000_000)




    def mark_image(self, image_index: int) -> None:
        """Associate this swapchain image with the current frame's fence."""
        self.images_in_flight[image_index] = self.fences[self.current_frame]




    def advance(self) -> None:
        self.current_frame = (self.current_frame + 1) % self.frame_count




    def destroy(self) -> None:
        for sem in self.render_finished:
            vk.vkDestroySemaphore(self.device, sem, None)
        for i in range(self.frame_count):
            vk.vkDestroySemaphore(self.device, self.image_available[i], None)
            vk.vkDestroyFence(self.device, self.fences[i], None)
        self.fences.clear()
        self.image_available.clear()
        self.render_finished.clear()
        self.images_in_flight.clear()