acdream/src/AcDream.App/Streaming/StreamingWorkBudget.cs

using System.Diagnostics;

namespace AcDream.App.Streaming;

/// <summary>
/// Immutable, validated per-frame streaming work profile.
/// </summary>
public readonly record struct StreamingWorkBudget
{
    public StreamingWorkBudget(
        TimeSpan maxUpdateTime,
        int maxCompletionAdmissions,
        long maxAdoptedCpuBytes,
        int maxEntityOperations,
        long maxGpuUploadBytes,
        int maxGlRetireOperations,
        float destinationReserveFraction)
    {
        if (maxUpdateTime <= TimeSpan.Zero)
            throw new ArgumentOutOfRangeException(nameof(maxUpdateTime));
        if (maxCompletionAdmissions <= 0)
            throw new ArgumentOutOfRangeException(nameof(maxCompletionAdmissions));
        if (maxAdoptedCpuBytes <= 0)
            throw new ArgumentOutOfRangeException(nameof(maxAdoptedCpuBytes));
        if (maxEntityOperations <= 0)
            throw new ArgumentOutOfRangeException(nameof(maxEntityOperations));
        if (maxGpuUploadBytes <= 0)
            throw new ArgumentOutOfRangeException(nameof(maxGpuUploadBytes));
        if (maxGlRetireOperations <= 0)
            throw new ArgumentOutOfRangeException(nameof(maxGlRetireOperations));
        if (!float.IsFinite(destinationReserveFraction)
            || destinationReserveFraction <= 0f
            || destinationReserveFraction >= 1f)
        {
            throw new ArgumentOutOfRangeException(
                nameof(destinationReserveFraction));
        }

        MaxUpdateTime = maxUpdateTime;
        MaxCompletionAdmissions = maxCompletionAdmissions;
        MaxAdoptedCpuBytes = maxAdoptedCpuBytes;
        MaxEntityOperations = maxEntityOperations;
        MaxGpuUploadBytes = maxGpuUploadBytes;
        MaxGlRetireOperations = maxGlRetireOperations;
        DestinationReserveFraction = destinationReserveFraction;
    }

    public TimeSpan MaxUpdateTime { get; }
    public int MaxCompletionAdmissions { get; }
    public long MaxAdoptedCpuBytes { get; }
    public int MaxEntityOperations { get; }
    public long MaxGpuUploadBytes { get; }
    public int MaxGlRetireOperations { get; }
    public float DestinationReserveFraction { get; }

    internal void Validate()
    {
        if (MaxUpdateTime <= TimeSpan.Zero
            || MaxCompletionAdmissions <= 0
            || MaxAdoptedCpuBytes <= 0
            || MaxEntityOperations <= 0
            || MaxGpuUploadBytes <= 0
            || MaxGlRetireOperations <= 0
            || !float.IsFinite(DestinationReserveFraction)
            || DestinationReserveFraction <= 0f
            || DestinationReserveFraction >= 1f)
        {
            throw new ArgumentException(
                "Streaming work budget is not a valid complete profile.",
                nameof(StreamingWorkBudget));
        }
    }
}

/// <summary>
/// Conservative, known cost of one atomic streaming operation.
/// </summary>
public readonly record struct StreamingWorkCost(
    int CompletionAdmissions = 0,
    long AdoptedCpuBytes = 0,
    int EntityOperations = 0,
    long GpuUploadBytes = 0,
    int GlRetireOperations = 0)
{
    public bool IsZero =>
        CompletionAdmissions == 0
        && AdoptedCpuBytes == 0
        && EntityOperations == 0
        && GpuUploadBytes == 0
        && GlRetireOperations == 0;

    internal void Validate()
    {
        if (CompletionAdmissions < 0)
            throw new ArgumentOutOfRangeException(nameof(CompletionAdmissions));
        if (AdoptedCpuBytes < 0)
            throw new ArgumentOutOfRangeException(nameof(AdoptedCpuBytes));
        if (EntityOperations < 0)
            throw new ArgumentOutOfRangeException(nameof(EntityOperations));
        if (GpuUploadBytes < 0)
            throw new ArgumentOutOfRangeException(nameof(GpuUploadBytes));
        if (GlRetireOperations < 0)
            throw new ArgumentOutOfRangeException(nameof(GlRetireOperations));
    }

    internal StreamingWorkCost Add(StreamingWorkCost other) => new(
        SaturatingAdd(CompletionAdmissions, other.CompletionAdmissions),
        SaturatingAdd(AdoptedCpuBytes, other.AdoptedCpuBytes),
        SaturatingAdd(EntityOperations, other.EntityOperations),
        SaturatingAdd(GpuUploadBytes, other.GpuUploadBytes),
        SaturatingAdd(GlRetireOperations, other.GlRetireOperations));

    private static int SaturatingAdd(int left, int right) =>
        left > int.MaxValue - right ? int.MaxValue : left + right;

    private static long SaturatingAdd(long left, long right) =>
        left > long.MaxValue - right ? long.MaxValue : left + right;
}

public enum StreamingWorkAdmission : byte
{
    Admitted,
    OversizedProgress,
    Yielded,
}

public enum StreamingWorkLimit : byte
{
    None,
    Time,
    CompletionAdmissions,
    AdoptedCpuBytes,
    EntityOperations,
    GpuUploadBytes,
    GlRetireOperations,
}

/// <summary>
/// Immutable observation of one frame's streaming work.
/// </summary>
public readonly record struct StreamingWorkMeterSnapshot(
    StreamingWorkCost Used,
    int Operations,
    int CompletedOperations,
    int YieldCount,
    int OverrunCount,
    int OversizedProgressCount,
    int FailureCount,
    double ElapsedMilliseconds,
    string? LastStage,
    StreamingWorkLimit LastLimit);

/// <summary>
/// Read-only streaming scheduler facts published to lifecycle artifacts.
/// </summary>
public readonly record struct StreamingWorkDiagnostics(
    StreamingWorkMeterSnapshot LastFrame,
    int DeferredCompletions,
    long DeferredAdoptedCpuBytes,
    double OldestDeferredAgeMilliseconds,
    int PendingPublications,
    int PendingRetirements);

/// <summary>
/// Single-thread, frame-scoped admission meter. Callers reserve a known cost
/// before an atomic operation and complete or fail that reservation afterward.
/// </summary>
public sealed class StreamingWorkMeter
{
    private readonly StreamingWorkBudget _budget;
    private readonly Func<long> _timestamp;
    private readonly long _frequency;
    private readonly long _start;
    private StreamingWorkCost _used;
    private int _operations;
    private int _completed;
    private int _yields;
    private int _overruns;
    private int _oversizedProgress;
    private int _failures;
    private bool _frameOverrunRecorded;
    private bool _reservationActive;
    private string? _activeStage;
    private string? _lastStage;
    private StreamingWorkLimit _lastLimit;

    public StreamingWorkMeter(StreamingWorkBudget budget)
        : this(budget, Stopwatch.GetTimestamp, Stopwatch.Frequency)
    {
    }

    public StreamingWorkMeter(
        StreamingWorkBudget budget,
        Func<long> timestamp,
        long timestampFrequency)
    {
        ArgumentNullException.ThrowIfNull(timestamp);
        if (timestampFrequency <= 0)
            throw new ArgumentOutOfRangeException(nameof(timestampFrequency));
        budget.Validate();

        _budget = budget;
        _timestamp = timestamp;
        _frequency = timestampFrequency;
        _start = timestamp();
    }

    public StreamingWorkAdmission TryReserve(
        StreamingWorkCost cost,
        string stage)
    {
        cost.Validate();
        ArgumentException.ThrowIfNullOrWhiteSpace(stage);
        if (_reservationActive)
            throw new InvalidOperationException(
                "The prior streaming operation has not completed.");

        StreamingWorkLimit limit = FindLimit(cost);
        if (limit != StreamingWorkLimit.None && _operations != 0)
        {
            _yields++;
            _lastStage = stage;
            _lastLimit = limit;
            return StreamingWorkAdmission.Yielded;
        }

        _used = _used.Add(cost);
        _operations++;
        _reservationActive = true;
        _activeStage = stage;
        _lastStage = stage;
        _lastLimit = limit;
        if (limit != StreamingWorkLimit.None)
        {
            _oversizedProgress++;
            return StreamingWorkAdmission.OversizedProgress;
        }

        return StreamingWorkAdmission.Admitted;
    }

    /// <summary>
    /// E1 observation seam: records work the legacy scheduler already chose to
    /// execute, while naming the budget decision that would have yielded. This
    /// preserves execution until the explicit scheduler takes ownership in E2.
    /// </summary>
    public void ForceReserveObserved(
        StreamingWorkCost cost,
        string stage)
    {
        cost.Validate();
        ArgumentException.ThrowIfNullOrWhiteSpace(stage);
        if (_reservationActive)
            throw new InvalidOperationException(
                "The prior streaming operation has not completed.");

        _used = _used.Add(cost);
        _operations++;
        _reservationActive = true;
        _activeStage = stage;
        _lastStage = stage;
    }

    public void Complete()
    {
        EnsureReservation();
        _completed++;
        _reservationActive = false;
        _activeStage = null;
        ObserveElapsedOverrun();
    }

    public void Fail()
    {
        EnsureReservation();
        _failures++;
        _reservationActive = false;
        _activeStage = null;
        ObserveElapsedOverrun();
    }

    public void FinishFrame()
    {
        if (_reservationActive)
            throw new InvalidOperationException(
                "A streaming operation is still reserved at frame end.");
        ObserveElapsedOverrun();
    }

    public StreamingWorkMeterSnapshot Snapshot
    {
        get
        {
            long now = _timestamp();
            return new StreamingWorkMeterSnapshot(
                _used,
                _operations,
                _completed,
                _yields,
                _overruns,
                _oversizedProgress,
                _failures,
                ElapsedMilliseconds(now),
                _lastStage,
                _lastLimit);
        }
    }

    private StreamingWorkLimit FindLimit(StreamingWorkCost cost)
    {
        if (ElapsedTicks(_timestamp()) >= _budget.MaxUpdateTime.Ticks)
            return StreamingWorkLimit.Time;
        if ((long)_used.CompletionAdmissions + cost.CompletionAdmissions
            > _budget.MaxCompletionAdmissions)
        {
            return StreamingWorkLimit.CompletionAdmissions;
        }
        if (_used.AdoptedCpuBytes > _budget.MaxAdoptedCpuBytes - cost.AdoptedCpuBytes)
            return StreamingWorkLimit.AdoptedCpuBytes;
        if ((long)_used.EntityOperations + cost.EntityOperations
            > _budget.MaxEntityOperations)
        {
            return StreamingWorkLimit.EntityOperations;
        }
        if (_used.GpuUploadBytes > _budget.MaxGpuUploadBytes - cost.GpuUploadBytes)
            return StreamingWorkLimit.GpuUploadBytes;
        if ((long)_used.GlRetireOperations + cost.GlRetireOperations
            > _budget.MaxGlRetireOperations)
        {
            return StreamingWorkLimit.GlRetireOperations;
        }
        return StreamingWorkLimit.None;
    }

    private void EnsureReservation()
    {
        if (!_reservationActive)
            throw new InvalidOperationException(
                "No streaming operation is reserved.");
    }

    private void ObserveElapsedOverrun()
    {
        if (!_frameOverrunRecorded
            && ElapsedTicks(_timestamp()) > _budget.MaxUpdateTime.Ticks)
        {
            _frameOverrunRecorded = true;
            _overruns++;
            _lastLimit = StreamingWorkLimit.Time;
            _lastStage = _activeStage ?? _lastStage;
        }
    }

    private long ElapsedTicks(long timestamp)
    {
        long raw = Math.Max(0L, timestamp - _start);
        return raw > long.MaxValue / TimeSpan.TicksPerSecond
            ? long.MaxValue
            : raw * TimeSpan.TicksPerSecond / _frequency;
    }

    private double ElapsedMilliseconds(long timestamp) =>
        Math.Max(0L, timestamp - _start) * 1000.0 / _frequency;
}