erik/acdream
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions

perf(streaming): reserve destination reveal capacity

Browse source 
Join destination scheduling to the canonical reveal generation, protect its share across every typed frame-budget dimension, and prevent stale work from clearing a replacement reservation. Remove forced incomplete materialization and project retail's centered portal wait cue while the authored tunnel remains active.

Tests: Release build clean; 91 focused reservation/reveal tests; full solution 8,158 passed, 5 skipped.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Erik 2026-07-24 19:39:23 +02:00
parent 98f1ac8934
commit 2ff8f844b0
33 changed files with 870 additions and 230 deletions
Download patch file Download diff file Expand all files Collapse all files

167
src/AcDream.App/Streaming/StreamingWorkBudget.cs
View file

@ -135,11 +135,19 @@ public enum StreamingWorkLimit : byte
GlRetireOperations,
}
internal enum StreamingWorkLane : byte
{
NonDestination,
Destination,
}
/// <summary>
/// Immutable observation of one frame's streaming work.
/// </summary>
public readonly record struct StreamingWorkMeterSnapshot(
StreamingWorkCost Used,
StreamingWorkCost DestinationUsed,
StreamingWorkCost NonDestinationUsed,
int Operations,
int CompletedOperations,
int YieldCount,
@ -177,7 +185,14 @@ public sealed class StreamingWorkMeter
private readonly Func<long> _timestamp;
private readonly long _frequency;
private readonly long _start;
private readonly bool _destinationReservationActive;
private StreamingWorkCost _used;
private StreamingWorkCost _destinationUsed;
private StreamingWorkCost _nonDestinationUsed;
private StreamingWorkLane _lane;
private long _destinationElapsedTicks;
private long _nonDestinationElapsedTicks;
private long _activeOperationStart;
private int _operations;
private int _completed;
private int _yields;
@ -191,15 +206,22 @@ public sealed class StreamingWorkMeter
private string? _lastStage;
private StreamingWorkLimit _lastLimit;
public StreamingWorkMeter(StreamingWorkBudget budget)
: this(budget, Stopwatch.GetTimestamp, Stopwatch.Frequency)
public StreamingWorkMeter(
StreamingWorkBudget budget,
bool destinationReservationActive = false)
: this(
budget,
Stopwatch.GetTimestamp,
Stopwatch.Frequency,
destinationReservationActive)
{
}
public StreamingWorkMeter(
StreamingWorkBudget budget,
Func<long> timestamp,
long timestampFrequency)
long timestampFrequency,
bool destinationReservationActive = false)
{
ArgumentNullException.ThrowIfNull(timestamp);
if (timestampFrequency <= 0)
@ -210,6 +232,18 @@ public sealed class StreamingWorkMeter
_timestamp = timestamp;
_frequency = timestampFrequency;
_start = timestamp();
_destinationReservationActive = destinationReservationActive;
}
internal LaneScope EnterLane(StreamingWorkLane lane)
{
if (_reservationActive)
throw new InvalidOperationException(
"Cannot change streaming lane during an active operation.");
StreamingWorkLane previous = _lane;
_lane = lane;
return new LaneScope(this, previous);
}
public StreamingWorkAdmission TryReserve(
@ -237,9 +271,14 @@ public sealed class StreamingWorkMeter
}
_used = _used.Add(cost);
if (_lane == StreamingWorkLane.Destination)
_destinationUsed = _destinationUsed.Add(cost);
else
_nonDestinationUsed = _nonDestinationUsed.Add(cost);
_operations++;
_reservationActive = true;
_activeStage = stage;
_activeOperationStart = _timestamp();
if (ensureProgress)
_ensuredProgressGranted = true;
if (limit != StreamingWorkLimit.None)
@ -258,31 +297,10 @@ public sealed class StreamingWorkMeter
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();
CompleteLaneTiming();
_completed++;
_reservationActive = false;
_activeStage = null;
@ -292,6 +310,7 @@ public sealed class StreamingWorkMeter
public void Fail()
{
EnsureReservation();
CompleteLaneTiming();
_failures++;
_reservationActive = false;
_activeStage = null;
@ -313,6 +332,8 @@ public sealed class StreamingWorkMeter
long now = _timestamp();
return new StreamingWorkMeterSnapshot(
_used,
_destinationUsed,
_nonDestinationUsed,
_operations,
_completed,
_yields,
@ -348,9 +369,105 @@ public sealed class StreamingWorkMeter
{
return StreamingWorkLimit.GlRetireOperations;
}
if (_destinationReservationActive
&& _lane == StreamingWorkLane.NonDestination)
{
double unreservedFraction =
1.0 - _budget.DestinationReserveFraction;
long maxTimeTicks = ReservedLimit(
_budget.MaxUpdateTime.Ticks,
unreservedFraction);
if (_nonDestinationElapsedTicks >= maxTimeTicks)
return StreamingWorkLimit.Time;
if ((long)_nonDestinationUsed.CompletionAdmissions
+ cost.CompletionAdmissions
> ReservedLimit(
_budget.MaxCompletionAdmissions,
unreservedFraction))
{
return StreamingWorkLimit.CompletionAdmissions;
}
if (_nonDestinationUsed.AdoptedCpuBytes
> ReservedLimit(
_budget.MaxAdoptedCpuBytes,
unreservedFraction) - cost.AdoptedCpuBytes)
{
return StreamingWorkLimit.AdoptedCpuBytes;
}
if ((long)_nonDestinationUsed.EntityOperations
+ cost.EntityOperations
> ReservedLimit(
_budget.MaxEntityOperations,
unreservedFraction))
{
return StreamingWorkLimit.EntityOperations;
}
if (_nonDestinationUsed.GpuUploadBytes
> ReservedLimit(
_budget.MaxGpuUploadBytes,
unreservedFraction) - cost.GpuUploadBytes)
{
return StreamingWorkLimit.GpuUploadBytes;
}
if ((long)_nonDestinationUsed.GlRetireOperations
+ cost.GlRetireOperations
> ReservedLimit(
_budget.MaxGlRetireOperations,
unreservedFraction))
{
return StreamingWorkLimit.GlRetireOperations;
}
}
return StreamingWorkLimit.None;
}
private static long ReservedLimit(long total, double fraction) =>
Math.Max(1L, (long)Math.Floor(total * fraction));
private void CompleteLaneTiming()
{
long raw = Math.Max(0L, _timestamp() - _activeOperationStart);
long elapsed = raw > long.MaxValue / TimeSpan.TicksPerSecond
? long.MaxValue
: raw * TimeSpan.TicksPerSecond / _frequency;
if (_lane == StreamingWorkLane.Destination)
_destinationElapsedTicks = SaturatingAdd(
_destinationElapsedTicks,
elapsed);
else
_nonDestinationElapsedTicks = SaturatingAdd(
_nonDestinationElapsedTicks,
elapsed);
}
private static long SaturatingAdd(long left, long right) =>
left > long.MaxValue - right ? long.MaxValue : left + right;
private void RestoreLane(StreamingWorkLane lane)
{
if (_reservationActive)
throw new InvalidOperationException(
"Cannot restore streaming lane during an active operation.");
_lane = lane;
}
internal readonly struct LaneScope : IDisposable
{
private readonly StreamingWorkMeter _meter;
private readonly StreamingWorkLane _previous;
internal LaneScope(
StreamingWorkMeter meter,
StreamingWorkLane previous)
{
_meter = meter;
_previous = previous;
}
public void Dispose() => _meter.RestoreLane(_previous);
}
private void EnsureReservation()
{
if (!_reservationActive)