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

perf(streaming): quiesce retired generations and budget teardown

Browse source 
Publish the retail blocking-for-cells edge before deferred recenter work, freeze old-world presentation/simulation/audio, and advance full-window retirement from exact metered entity and owner cursors. This removes synchronous portal teardown without allowing retained owners to remain observable.
This commit is contained in:
Erik 2026-07-24 18:29:52 +02:00
parent b8f6317fe1
commit bb16f74fd4
38 changed files with 1691 additions and 170 deletions
Download patch file Download diff file Expand all files Collapse all files

441
src/AcDream.App/Streaming/StreamingController.cs
View file

@ -29,11 +29,24 @@ public sealed class StreamingController : IStreamingFrameBackend
public List<uint>? ResidentIds;
public int RetirementCursor;
public bool PreparationCommitted;
public IEnumerator<uint>? ResidentEnumerator;
public (int X, int Y, bool IsSealedDungeon)? Destination;
public bool DestinationConfigured;
public bool DestinationLoadEnqueued;
}
private sealed class FullWindowRetirement
{
public bool GenerationAdvanced;
public bool PendingLoadsCleared;
public bool CompletionQueueCleared;
public bool RegionCleared;
public List<uint>? ResidentIds;
public IEnumerator<uint>? ResidentEnumerator;
public int RetirementCursor;
public bool PreparationCommitted;
}
private readonly Action<uint, LandblockStreamJobKind, ulong> _enqueueLoad;
private readonly Action<uint, ulong> _enqueueUnload;
private readonly ILandblockCompletionSource _completionSource;
@ -55,6 +68,7 @@ public sealed class StreamingController : IStreamingFrameBackend
private bool _advancingRadiiReconfiguration;
private (int NearRadius, int FarRadius)? _deferredRadiiRequest;
private OriginRecenterRetirement? _originRecenterRetirement;
private FullWindowRetirement? _fullWindowRetirement;
private bool _advancingOriginRecenter;
// Hard streaming boundaries advance this token. Worker completions carry
// the generation captured at enqueue time, so an old overlapping load or
@ -459,7 +473,7 @@ public sealed class StreamingController : IStreamingFrameBackend
private void DemoteLandblock(uint id)
{
uint canonical = (id & 0xFFFF0000u) | 0xFFFFu;
_presentation.BeginNearLayerRetirement(canonical);
_presentation.EnqueueNearLayerRetirement(canonical);
}
private bool AdvanceGeneration()
@ -530,9 +544,37 @@ public sealed class StreamingController : IStreamingFrameBackend
_activeWorkMeter = meter;
try
{
if (_fullWindowRetirement is not null)
{
bool retirementWasPending =
_presentation.PendingRetirementCount != 0;
TryAdvanceFullWindowRetirement(meter);
if (_presentation.UsesBudgetedRetirementSteps
|| retirementWasPending)
{
_presentation.AdvanceRetirements(meter);
}
if (_fullWindowRetirement is
{
PreparationCommitted: true,
}
&& _presentation.PendingRetirementCount == 0)
{
_fullWindowRetirement = null;
}
return;
}
if (_originRecenterRetirement is not null)
{
_presentation.AdvanceRetirements();
bool retirementWasPending =
_presentation.PendingRetirementCount != 0;
TryAdvanceOriginRecenterPreparation(meter);
if (_presentation.UsesBudgetedRetirementSteps
|| retirementWasPending)
{
_presentation.AdvanceRetirements(meter);
}
return;
}
@ -544,12 +586,12 @@ public sealed class StreamingController : IStreamingFrameBackend
ReconfigureRadii(deferred.NearRadius, deferred.FarRadius);
}
// Presentation-owner failures never roll back spatial state. Resume
// unfinished owner ledgers before accepting replacement publications.
_presentation.AdvanceRetirements();
// Complete an admitted publication before this frame can mutate its
// desired tier or spatial residence. A later recenter may then demote
// or retire the fully known owner set through the normal ledger.
bool retirementWasPendingAtFrameStart =
_presentation.PendingRetirementCount != 0;
_presentation.AdvanceRetirements(meter);
if (!ConvergePendingPublications())
return;
@ -584,6 +626,12 @@ public sealed class StreamingController : IStreamingFrameBackend
}
DrainAndApply();
// Retirement is cleanup after immediate spatial withdrawal. Spend
// remaining capacity only after visible/destination publication
// work has had access to this frame's shared meter.
if (_presentation.UsesBudgetedRetirementSteps
&& !retirementWasPendingAtFrameStart)
_presentation.AdvanceRetirements(meter);
}
finally
{
@ -882,10 +930,13 @@ public sealed class StreamingController : IStreamingFrameBackend
}
/// <summary>
/// Starts a reload of the current streaming window. Logical residency is
/// detached immediately; render, physics, and static-lighting owners then
/// converge through the retained presentation-retirement ledger before
/// <see cref="NormalTick"/> bootstraps the window again.
/// Starts a reload of the current streaming window. The next and later
/// <see cref="Tick"/> calls invalidate accepted work, capture and detach
/// stable residents, and converge render, physics, and static-lighting
/// owners under the shared frame budget before <see cref="NormalTick"/>
/// bootstraps the window again. Hard login/portal transitions separately
/// publish their immediate observable edge through
/// <see cref="WorldGenerationQuiescence"/>.
/// Shared-origin teleports use <see cref="BeginOriginRecenter"/> instead so
/// every old-window owner retires while the old coordinate frame is active.
/// </summary>
@ -916,10 +967,8 @@ public sealed class StreamingController : IStreamingFrameBackend
throw new InvalidOperationException(
"No streaming-origin recenter transaction is pending.");
if (!TryAdvanceOriginRecenterPreparation())
return false;
_presentation.AdvanceRetirements();
return _presentation.PendingRetirementCount == 0;
return _originRecenterRetirement.PreparationCommitted
&& _presentation.PendingRetirementCount == 0;
}
/// <summary>
@ -1040,8 +1089,9 @@ public sealed class StreamingController : IStreamingFrameBackend
return true;
}
private bool TryAdvanceOriginRecenterPreparation()
private bool TryAdvanceOriginRecenterPreparation(StreamingWorkMeter meter)
{
ArgumentNullException.ThrowIfNull(meter);
OriginRecenterRetirement transaction = _originRecenterRetirement
?? throw new InvalidOperationException(
"No streaming-origin recenter transaction is pending.");
@ -1055,58 +1105,159 @@ public sealed class StreamingController : IStreamingFrameBackend
{
if (!transaction.RadiiConverged)
{
if (_pendingRadiiReconfiguration is not null)
AdvanceRadiiReconfiguration();
if (_pendingRadiiReconfiguration is not null
&& !TryRunStreamingWork(
meter,
new StreamingWorkCost(
EntityOperations: Math.Max(
1,
_pendingRadiiReconfiguration.Mutations.Count)),
"recenter-radii-convergence",
() =>
{
AdvanceRadiiReconfiguration();
return _pendingRadiiReconfiguration is null;
}))
{
return false;
}
if (_pendingRadiiReconfiguration is not null)
return false;
transaction.RadiiConverged = true;
}
if (!transaction.GenerationAdvanced)
{
// AdvanceGeneration meters any retained publication it must
// converge through the active frame meter. The token increment
// itself is constant-time and must not create a nested meter
// reservation around that retry.
if (!AdvanceGeneration())
return false;
transaction.GenerationAdvanced = true;
}
if (!transaction.PendingLoadsCleared)
{
try
bool ClearPendingLoads()
{
_clearPendingLoads?.Invoke();
transaction.PendingLoadsCleared = true;
try
{
_clearPendingLoads?.Invoke();
transaction.PendingLoadsCleared = true;
return true;
}
catch (StreamingMutationException error) when (error.MutationCommitted)
{
transaction.PendingLoadsCleared = true;
Console.WriteLine(
$"streaming: committed pending-load clear reported failure: {error}");
return false;
}
}
catch (StreamingMutationException error) when (error.MutationCommitted)
{
transaction.PendingLoadsCleared = true;
Console.WriteLine(
$"streaming: committed pending-load clear reported failure: {error}");
if (!TryRunStreamingWork(
meter,
new StreamingWorkCost(EntityOperations: 1),
"recenter-clear-worker-inbox",
ClearPendingLoads))
return false;
}
}
if (!transaction.CompletionQueueCleared)
{
_completionQueue.Clear();
while (_completionQueue.Count != 0)
{
StreamingWorkAdmission admission = meter.TryReserve(
new StreamingWorkCost(EntityOperations: 1),
"recenter-release-completion");
if (admission == StreamingWorkAdmission.Yielded)
return false;
if (!_completionQueue.TryRemoveOne())
{
meter.Fail();
throw new InvalidOperationException(
"Completion queue count changed during recenter release.");
}
meter.Complete();
}
transaction.CompletionQueueCleared = true;
}
if (!transaction.RegionCleared)
{
_collapsed = false;
_region = null;
if (!TryRunStreamingWork(
meter,
new StreamingWorkCost(EntityOperations: 1),
"recenter-clear-region",
() =>
{
_collapsed = false;
_region = null;
return true;
}))
{
return false;
}
transaction.RegionCleared = true;
}
if (transaction.ResidentIds is null)
{
transaction.ResidentIds = new List<uint>(_state.LoadedLandblockIds);
FullWindowRetirementCount++;
LastFullWindowRetirementLandblockCount = transaction.ResidentIds.Count;
transaction.ResidentIds = [];
transaction.ResidentEnumerator =
_state.LoadedLandblockIds.GetEnumerator();
}
while (transaction.RetirementCursor < transaction.ResidentIds.Count)
while (transaction.ResidentEnumerator is { } residentEnumerator)
{
uint id = transaction.ResidentIds[transaction.RetirementCursor];
StreamingWorkAdmission admission = meter.TryReserve(
new StreamingWorkCost(EntityOperations: 1),
"recenter-capture-resident-id");
if (admission == StreamingWorkAdmission.Yielded)
return false;
bool moved;
try
{
_presentation.BeginFullRetirement(id);
moved = residentEnumerator.MoveNext();
if (moved)
transaction.ResidentIds.Add(residentEnumerator.Current);
else
{
residentEnumerator.Dispose();
transaction.ResidentEnumerator = null;
FullWindowRetirementCount++;
LastFullWindowRetirementLandblockCount =
transaction.ResidentIds.Count;
}
meter.Complete();
}
catch
{
meter.Fail();
throw;
}
}
List<uint> residentIds = transaction.ResidentIds
?? throw new InvalidOperationException(
"Origin-recenter resident capture did not commit.");
while (transaction.RetirementCursor < residentIds.Count)
{
uint id = residentIds[transaction.RetirementCursor];
int entityCount = _state.TryGetLandblock(
id,
out LoadedLandblock? loaded)
? loaded!.Entities.Count
: 0;
StreamingWorkAdmission admission = meter.TryReserve(
new StreamingWorkCost(
EntityOperations: Math.Max(1, entityCount)),
$"recenter-detach-0x{id:X8}");
if (admission == StreamingWorkAdmission.Yielded)
return false;
try
{
_presentation.EnqueueFullRetirement(id);
transaction.RetirementCursor++;
meter.Complete();
}
catch (Exception error)
{
@ -1116,6 +1267,7 @@ public sealed class StreamingController : IStreamingFrameBackend
// otherwise retain it for the next frame.
if (!_state.IsLoaded(id))
transaction.RetirementCursor++;
meter.Fail();
Console.WriteLine(
$"streaming: origin-recenter retirement for 0x{id:X8} " +
$"will resume: {error}");
@ -1140,21 +1292,178 @@ public sealed class StreamingController : IStreamingFrameBackend
private void BeginFullWindowRetirement()
{
if (!AdvanceGeneration())
return;
_collapsed = false;
_clearPendingLoads?.Invoke();
_completionQueue.Clear();
// Commit the old region boundary before any presentation owner is
// advanced. New same-id publications are fenced by the presentation
// pipeline's retryable retirement ledger.
_region = null;
// Snapshot — RemoveLandblock mutates the loaded set we're iterating.
var ids = new List<uint>(_state.LoadedLandblockIds);
FullWindowRetirementCount++; // [FRAME-DIAG] churn counter
LastFullWindowRetirementLandblockCount = ids.Count; // upcoming re-upload volume
foreach (var id in ids)
_presentation.BeginFullRetirement(id);
_fullWindowRetirement ??= new FullWindowRetirement();
}
private bool TryAdvanceFullWindowRetirement(StreamingWorkMeter meter)
{
FullWindowRetirement transaction = _fullWindowRetirement
?? throw new InvalidOperationException(
"No full-window retirement transaction is pending.");
if (transaction.PreparationCommitted)
return true;
try
{
if (!transaction.GenerationAdvanced)
{
if (!AdvanceGeneration())
{
return false;
}
transaction.GenerationAdvanced = true;
}
if (!transaction.PendingLoadsCleared)
{
bool ClearPendingLoads()
{
try
{
_clearPendingLoads?.Invoke();
transaction.PendingLoadsCleared = true;
return true;
}
catch (StreamingMutationException error) when (error.MutationCommitted)
{
transaction.PendingLoadsCleared = true;
Console.WriteLine(
$"streaming: committed reload pending-load clear reported failure: {error}");
return false;
}
}
if (!TryRunStreamingWork(
meter,
new StreamingWorkCost(EntityOperations: 1),
"reload-clear-worker-inbox",
ClearPendingLoads))
{
return false;
}
}
if (!transaction.CompletionQueueCleared)
{
while (_completionQueue.Count != 0)
{
StreamingWorkAdmission admission = meter.TryReserve(
new StreamingWorkCost(EntityOperations: 1),
"reload-release-completion");
if (admission == StreamingWorkAdmission.Yielded)
return false;
if (!_completionQueue.TryRemoveOne())
{
meter.Fail();
throw new InvalidOperationException(
"Completion queue count changed during reload release.");
}
meter.Complete();
}
transaction.CompletionQueueCleared = true;
}
if (!transaction.RegionCleared)
{
if (!TryRunStreamingWork(
meter,
new StreamingWorkCost(EntityOperations: 1),
"reload-clear-region",
() =>
{
_collapsed = false;
_region = null;
return true;
}))
{
return false;
}
transaction.RegionCleared = true;
}
if (transaction.ResidentIds is null)
{
transaction.ResidentIds = [];
transaction.ResidentEnumerator =
_state.LoadedLandblockIds.GetEnumerator();
}
while (transaction.ResidentEnumerator is { } residentEnumerator)
{
StreamingWorkAdmission admission = meter.TryReserve(
new StreamingWorkCost(EntityOperations: 1),
"reload-capture-resident-id");
if (admission == StreamingWorkAdmission.Yielded)
return false;
bool moved;
try
{
moved = residentEnumerator.MoveNext();
if (moved)
transaction.ResidentIds.Add(residentEnumerator.Current);
else
{
residentEnumerator.Dispose();
transaction.ResidentEnumerator = null;
FullWindowRetirementCount++;
LastFullWindowRetirementLandblockCount =
transaction.ResidentIds.Count;
}
meter.Complete();
}
catch
{
meter.Fail();
throw;
}
}
List<uint> residentIds = transaction.ResidentIds
?? throw new InvalidOperationException(
"Full-window resident capture did not commit.");
while (transaction.RetirementCursor < residentIds.Count)
{
uint id = residentIds[transaction.RetirementCursor];
int entityCount = _state.TryGetLandblock(
id,
out LoadedLandblock? loaded)
? loaded!.Entities.Count
: 0;
StreamingWorkAdmission admission = meter.TryReserve(
new StreamingWorkCost(
EntityOperations: Math.Max(1, entityCount)),
$"reload-detach-0x{id:X8}");
if (admission == StreamingWorkAdmission.Yielded)
return false;
try
{
_presentation.EnqueueFullRetirement(id);
transaction.RetirementCursor++;
meter.Complete();
}
catch (Exception error)
{
if (!_state.IsLoaded(id))
transaction.RetirementCursor++;
meter.Fail();
Console.WriteLine(
$"streaming: full-window retirement for 0x{id:X8} " +
$"will resume: {error}");
return false;
}
}
transaction.PreparationCommitted = true;
return true;
}
catch (Exception error)
{
Console.WriteLine(
$"streaming: full-window preparation will resume: {error}");
return false;
}
}
/// <summary>
@ -1209,6 +1518,32 @@ public sealed class StreamingController : IStreamingFrameBackend
}
}
private static bool TryRunStreamingWork(
StreamingWorkMeter meter,
StreamingWorkCost cost,
string stage,
Func<bool> operation)
{
StreamingWorkAdmission admission = meter.TryReserve(cost, stage);
if (admission == StreamingWorkAdmission.Yielded)
return false;
try
{
bool completed = operation();
if (completed)
meter.Complete();
else
meter.Fail();
return completed;
}
catch
{
meter.Fail();
throw;
}
}
private void AdmitCompletions(StreamingWorkMeter meter)
{
while (_completionSource.TryPeek(out LandblockStreamResult? peeked))
@ -1439,7 +1774,7 @@ public sealed class StreamingController : IStreamingFrameBackend
mergeIntoExistingLandblock: _state.IsLoaded(promoted.LandblockId));
break;
case LandblockStreamResult.Unloaded unloaded:
_presentation.BeginFullRetirement(unloaded.LandblockId);
_presentation.EnqueueFullRetirement(unloaded.LandblockId);
break;
case LandblockStreamResult.Failed failed:
Console.WriteLine(
@ -1458,14 +1793,14 @@ public sealed class StreamingController : IStreamingFrameBackend
|| !_region.TryGetDesiredTier(landblockId, out LandblockStreamTier desiredTier))
{
if (_state.IsLoaded(landblockId))
_presentation.BeginFullRetirement(landblockId);
_presentation.EnqueueFullRetirement(landblockId);
return;
}
if (desiredTier == LandblockStreamTier.Far
&& _state.IsNearTier(landblockId))
{
_presentation.BeginNearLayerRetirement(landblockId);
_presentation.EnqueueNearLayerRetirement(landblockId);
}
}