java-topology/defects/sdl3/unit/test_sdl3_0001.py

"""
Unit test for sdl3-0001: SDL3 GPU TRACK_RESOURCE O(N²) linear dedup.

Tests:
1. Correctness: both implementations produce the same unique resource list
2. Performance: O(N²) linear scan vs O(N) hash set ratio
"""
import time
import random


# ---- Minimal resource stub ----

class FakeResource:
    """Simulates a VulkanTexture* or D3D12Texture* pointer."""
    def __init__(self, resource_id):
        self.resource_id = resource_id
        self.ref_count = 0


# ---- BEFORE: defective implementation (linear scan dedup) ----

def track_resource_before(used_list, resource):
    """
    Mirrors the TRACK_RESOURCE macro in SDL_gpu_vulkan.c and SDL_gpu_d3d12.c.
    O(N) linear scan for duplicate check before adding.
    """
    for r in reversed(used_list):   # Vulkan scans in reverse
        if r is resource:
            return  # duplicate, skip
    used_list.append(resource)
    resource.ref_count += 1


def record_frame_before(resources_per_draw, draw_count):
    """
    Simulate recording a frame: for each draw call, bind a set of resources.
    Returns the final tracked list.
    """
    used_textures = []
    for draw_i in range(draw_count):
        for resource in resources_per_draw[draw_i]:
            track_resource_before(used_textures, resource)
    return used_textures


# ---- AFTER: fixed implementation (hash set dedup) ----

def track_resource_after(used_set, used_list, resource):
    """
    O(1) hash set membership check.
    """
    resource_id = id(resource)
    if resource_id not in used_set:
        used_set.add(resource_id)
        used_list.append(resource)
        resource.ref_count += 1


def record_frame_after(resources_per_draw, draw_count):
    """
    Same frame recording but using hash set for dedup.
    """
    used_textures = []
    used_set = set()
    for draw_i in range(draw_count):
        for resource in resources_per_draw[draw_i]:
            track_resource_after(used_set, used_textures, resource)
    return used_textures


# ---- Tests ----

def test_correctness_no_duplicates():
    """All unique resources are tracked exactly once."""
    resources = [FakeResource(i) for i in range(50)]
    draw_schedule = [resources[i:i+5] for i in range(0, 50, 5)]
    result = record_frame_before(draw_schedule, len(draw_schedule))
    # All 50 resources should appear exactly once
    assert len(result) == 50, f"Expected 50, got {len(result)}"
    assert all(r.ref_count == 1 for r in resources), "Ref counts corrupted"
    print("PASS test_correctness_no_duplicates")


def test_correctness_with_duplicates():
    """Resources reused across draw calls are tracked only once."""
    r0 = FakeResource(0)
    r1 = FakeResource(1)
    r2 = FakeResource(2)
    # Same textures reused across many draw calls (typical render pass)
    draw_schedule = [[r0, r1, r2]] * 10
    result = record_frame_before(draw_schedule, len(draw_schedule))
    assert len(result) == 3, f"Expected 3 unique, got {len(result)}"
    assert r0.ref_count == 1 and r1.ref_count == 1 and r2.ref_count == 1
    print("PASS test_correctness_with_duplicates")


def test_correctness_before_after_agree():
    """Before and after produce identical resource lists."""
    n_resources = 100
    n_draws = 50
    resources = [FakeResource(i) for i in range(n_resources)]
    rng = random.Random(42)
    draw_schedule = [
        [resources[rng.randint(0, n_resources - 1)] for _ in range(5)]
        for _ in range(n_draws)
    ]

    # Reset ref counts
    for r in resources:
        r.ref_count = 0
    before_result = record_frame_before(draw_schedule, n_draws)

    for r in resources:
        r.ref_count = 0
    after_result = record_frame_after(draw_schedule, n_draws)

    assert set(id(r) for r in before_result) == set(id(r) for r in after_result), (
        "Before and after tracked different resource sets"
    )
    print("PASS test_correctness_before_after_agree")


def test_performance_typical_scene():
    """Worst-case: all N resources tracked in sequence (each new, no duplicates).
    This directly measures the O(N²) sum(1+2+...+N) vs O(N) hash cost."""
    N = 200      # unique textures bound across the command buffer
    ITERS = 2000

    resources = [FakeResource(i) for i in range(N)]
    # Each draw binds exactly one new resource — worst-case for linear scan
    draw_schedule = [[resources[i]] for i in range(N)]

    t0 = time.perf_counter()
    for _ in range(ITERS):
        for r in resources:
            r.ref_count = 0
        record_frame_before(draw_schedule, N)
    t_before = time.perf_counter() - t0

    t0 = time.perf_counter()
    for _ in range(ITERS):
        for r in resources:
            r.ref_count = 0
        record_frame_after(draw_schedule, N)
    t_after = time.perf_counter() - t0

    ratio = t_before / t_after if t_after > 0 else float('inf')
    print(f"PERF sdl3-0001 (typical): N={N} ITERS={ITERS} "
          f"before={t_before*1000:.1f}ms after={t_after*1000:.1f}ms ratio={ratio:.1f}x")
    assert ratio >= 5, f"Expected >= 5x speedup, got {ratio:.1f}x"
    print("PASS test_performance_typical_scene")


def test_performance_heavy_scene():
    """Heavy scene: 1000 unique resources tracked sequentially — pure O(N²) vs O(N)."""
    N = 1000     # unique GPU resources per command buffer
    ITERS = 500

    resources = [FakeResource(i) for i in range(N)]
    # Every resource is new — forces full O(N) scan each time
    draw_schedule = [[resources[i]] for i in range(N)]

    t0 = time.perf_counter()
    for _ in range(ITERS):
        for r in resources:
            r.ref_count = 0
        record_frame_before(draw_schedule, N)
    t_before = time.perf_counter() - t0

    t0 = time.perf_counter()
    for _ in range(ITERS):
        for r in resources:
            r.ref_count = 0
        record_frame_after(draw_schedule, N)
    t_after = time.perf_counter() - t0

    ratio = t_before / t_after if t_after > 0 else float('inf')
    print(f"PERF sdl3-0001 (heavy): N={N} ITERS={ITERS} "
          f"before={t_before*1000:.1f}ms after={t_after*1000:.1f}ms ratio={ratio:.1f}x")
    assert ratio >= 30, f"Expected >= 30x speedup for N={N}, got {ratio:.1f}x"
    print("PASS test_performance_heavy_scene")


def test_no_ref_count_double_increment():
    """A resource reused across 100 draws is ref-counted exactly once."""
    shared = FakeResource(0)
    other = [FakeResource(i + 1) for i in range(10)]
    draw_schedule = [[shared] + other[:3]] * 100
    for r in [shared] + other:
        r.ref_count = 0
    record_frame_before(draw_schedule, len(draw_schedule))
    assert shared.ref_count == 1, (
        f"shared.ref_count should be 1 (not {shared.ref_count}); "
        "double-tracking corrupts refcounts"
    )
    print("PASS test_no_ref_count_double_increment")


if __name__ == "__main__":
    random.seed(42)
    test_correctness_no_duplicates()
    test_correctness_with_duplicates()
    test_correctness_before_after_agree()
    test_performance_typical_scene()
    test_performance_heavy_scene()
    test_no_ref_count_double_increment()
    print("\nAll sdl3-0001 tests PASSED")