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

refactor(streaming): own landblock publication transactions

Browse source
This commit is contained in:
Erik 2026-07-21 19:55:24 +02:00
parent 4001251064
commit 4f965d0699
5 changed files with 1301 additions and 122 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -21,24 +21,9 @@ public sealed class StreamingController
private readonly Action<uint, LandblockStreamJobKind, ulong> _enqueueLoad;
private readonly Action<uint, ulong> _enqueueUnload;
private readonly Func<int, IReadOnlyList<LandblockStreamResult>> _drainCompletions;
private readonly Action<LandblockBuild, LandblockMeshData> _applyTerrain;
private readonly Action? _clearPendingLoads;
private readonly LandblockRetirementCoordinator _retirements;
private readonly LandblockPresentationPipeline _presentation;
/// <summary>
/// #138: fired after a landblock's entity layer (re)loads — both the full
/// <see cref="LandblockStreamResult.Loaded"/> path (initial / dungeon-exit
/// expand) and the <see cref="LandblockStreamResult.Promoted"/> path
/// (Far→Near). <c>GameWindow</c> wires it to re-hydrate server objects whose
/// render entities were dropped by the collapse/demote but whose parsed
/// spawns it still retains. Receives the canonical landblock id. Null in
/// tests that don't exercise re-hydration.
/// </summary>
private readonly Action<uint>? _onLandblockLoaded;
// Invoked only after LandblockSpawnAdapter has pinned every synthetic
// EnvCell geometry id. This re-arms the schema-aware preparation path if
// an unowned cached mesh was evicted between worker schedule and publish.
private readonly Action<EnvCellLandblockBuild>? _ensureEnvCellMeshes;
private readonly GpuWorldState _state;
private StreamingRegion? _region;
private RadiiReconfiguration? _pendingRadiiReconfiguration;
@ -194,16 +179,16 @@ public sealed class StreamingController
_enqueueLoad = enqueueLoad;
_enqueueUnload = enqueueUnload;
_drainCompletions = drainCompletions;
_applyTerrain = applyTerrain;
_clearPendingLoads = clearPendingLoads;
_onLandblockLoaded = onLandblockLoaded;
_ensureEnvCellMeshes = ensureEnvCellMeshes;
_state = state;
_retirements = retirementCoordinator
?? LandblockRetirementCoordinator.CreateLegacy(
state,
removeTerrain,
demoteNearLayer);
_presentation = new LandblockPresentationPipeline(
applyTerrain,
state,
onLandblockLoaded,
ensureEnvCellMeshes,
retirementCoordinator,
removeTerrain,
demoteNearLayer);
NearRadius = nearRadius;
FarRadius = farRadius;
}
@ -249,6 +234,12 @@ public sealed class StreamingController
if (nearRadius == NearRadius && farRadius == FarRadius)
return;
// A staged old-window publication must either finish or fail before
// the desired-tier snapshot below is computed. Otherwise a landblock
// that becomes spatially resident during convergence is absent from
// the reconfiguration mutation ledger.
ConvergePendingPublications();
if (_collapsed || _region is null)
{
NearRadius = nearRadius;
@ -348,14 +339,44 @@ public sealed class StreamingController
private void DemoteLandblock(uint id)
{
uint canonical = (id & 0xFFFF0000u) | 0xFFFFu;
_retirements.BeginNearLayer(canonical);
_presentation.BeginNearLayerRetirement(canonical);
}
private void AdvanceGeneration()
{
ConvergePendingPublications();
_generation = unchecked(_generation + 1);
}
private void ConvergePendingPublications()
{
IReadOnlyList<LandblockStreamResult> pending =
_presentation.GetPendingPublicationResults();
if (pending.Count == 0)
return;
try
{
for (int i = 0; i < pending.Count; i++)
_presentation.ResumePublication(pending[i]);
}
finally
{
// A completed staged result can still be present in the deferred
// FIFO that originally retained it. Remove only those exact
// objects; unrelated accepted completions keep their order.
_deferredApply.RemoveAll(result =>
{
for (int i = 0; i < pending.Count; i++)
{
if (ReferenceEquals(result, pending[i]))
return !_presentation.HasPendingPublication(result);
}
return false;
});
}
}
/// <summary>
/// Advance one frame. <paramref name="observerCx"/>/<paramref name="observerCy"/>
/// are landblock coordinates (0..255) of the current viewer — the camera
@ -382,7 +403,11 @@ public sealed class StreamingController
// Presentation-owner failures never roll back spatial state. Resume
// unfinished owner ledgers before accepting replacement publications.
_retirements.Advance();
_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.
ConvergePendingPublications();
uint centerId = StreamingRegion.EncodeLandblockId(observerCx, observerCy);
@ -616,11 +641,12 @@ public sealed class StreamingController
/// </summary>
private void EnterDungeonCollapse(int cx, int cy, uint centerId)
{
if (!_collapsed || _collapsedCenter != centerId)
bool logTransition = !_collapsed || _collapsedCenter != centerId;
if (logTransition)
Console.WriteLine($"streaming: dungeon collapse -> 0x{centerId:X8}");
AdvanceGeneration();
_collapsed = true;
_collapsedCenter = centerId;
AdvanceGeneration();
_clearPendingLoads?.Invoke();
_deferredApply.Clear();
_region = null;
@ -685,8 +711,8 @@ public sealed class StreamingController
Console.WriteLine(
$"streaming: dungeon EXIT-expand -> ({observerCx},{observerCy}) " +
$"(was collapsed on 0x{_collapsedCenter:X8})");
_collapsed = false;
AdvanceGeneration();
_collapsed = false;
var rebuilt = new StreamingRegion(observerCx, observerCy, NearRadius, FarRadius);
foreach (var id in _state.LoadedLandblockIds)
@ -715,19 +741,20 @@ public sealed class StreamingController
/// </summary>
public void ForceReloadWindow()
{
_collapsed = false;
AdvanceGeneration();
_collapsed = false;
_clearPendingLoads?.Invoke();
_deferredApply.Clear();
// Commit the old region boundary before any presentation owner is
// advanced. New same-id publications are fenced by _retirements.
// 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);
ForceReloadCount++; // [FRAME-DIAG] churn counter
LastForceReloadDropCount = ids.Count; // = upcoming re-upload volume
foreach (var id in ids)
_retirements.BeginFull(id);
_presentation.BeginFullRetirement(id);
}
/// <summary>
@ -757,18 +784,41 @@ public sealed class StreamingController
{
var chunk = _drainCompletions(MaxCompletionsPerFrame);
if (chunk.Count == 0) break;
foreach (var result in chunk)
for (int chunkIndex = 0; chunkIndex < chunk.Count; chunkIndex++)
{
LandblockStreamResult result = chunk[chunkIndex];
// Reject obsolete hard-window work before it can occupy the
// metered deferred queue. ApplyResult repeats this invariant
// for direct callers and future drain paths.
if (IsStaleGeneration(result))
if (IsStaleGeneration(result)
&& !_presentation.HasPendingPublication(result))
continue;
if (IsPublicationBlockedByRetirement(result))
_deferredApply.Add(result);
else if (result is LandblockStreamResult.Unloaded
|| IsWithinPriorityRing(ResultLandblockId(result)))
ApplyResult(result); // free (unload) or behind-the-fade near ring
{
try
{
ApplyResult(result); // free (unload) or behind-the-fade near ring
}
catch
{
// The presentation pipeline retains its exact committed
// stage after the coarse presentation callback commits.
// Preserve it and every untouched completion from this
// already-drained chunk in original FIFO order. A
// presentation-stage failure is deliberately fail-fast
// until the focused publishers replace that callback;
// its result is not retryable, but the untouched suffix
// must still survive.
if (_presentation.HasPendingPublication(result))
_deferredApply.Add(result);
for (int suffix = chunkIndex + 1; suffix < chunk.Count; suffix++)
_deferredApply.Add(chunk[suffix]);
throw;
}
}
else
_deferredApply.Add(result); // a GPU-upload load — meter it in step 2
}
@ -798,9 +848,24 @@ public sealed class StreamingController
// Advance read only after the apply commits. If it throws, the
// finally block removes prior committed entries but retains this
// exact result and the untouched tail for retry.
ApplyResult(result);
read++;
applied++;
try
{
ApplyResult(result);
read++;
applied++;
}
catch
{
// A suffix-stage failure keeps its pipeline receipt and
// must remain at this exact FIFO position. The still-
// monolithic presentation callback deliberately removes
// its receipt on failure because its internal committed
// prefix is unknown; consume only that failed result so a
// later frame cannot replay it automatically.
if (!_presentation.HasPendingPublication(result))
read++;
throw;
}
}
}
finally
@ -821,6 +886,13 @@ public sealed class StreamingController
/// </summary>
private void ApplyResult(LandblockStreamResult result)
{
if (_presentation.HasPendingPublication(result))
{
_presentation.ResumePublication(result);
ReconcileCompletedPendingPublication(result.LandblockId);
return;
}
if (IsStaleGeneration(result))
{
// A worker can finish one old load/unload after a hard recenter.
@ -853,7 +925,9 @@ public sealed class StreamingController
bool isNearCompletion = result is LandblockStreamResult.Promoted
or LandblockStreamResult.Loaded { Tier: LandblockStreamTier.Near };
if (isNearCompletion && _state.IsNearTier(result.LandblockId))
if (isNearCompletion
&& _state.IsNearTier(result.LandblockId)
&& !_presentation.HasPendingPublication(result))
{
// Normal recenter can enqueue a second high-priority Near job
// while an earlier one is already in flight. The streamer
@ -878,10 +952,10 @@ public sealed class StreamingController
switch (result)
{
case LandblockStreamResult.Loaded loaded:
ApplyAsFar(loaded.Build, loaded.MeshData);
_presentation.PublishAsFar(loaded, loaded.Build, loaded.MeshData);
break;
case LandblockStreamResult.Promoted promoted:
ApplyAsFar(promoted.Build, promoted.MeshData);
_presentation.PublishAsFar(promoted, promoted.Build, promoted.MeshData);
break;
}
}
@ -900,16 +974,7 @@ public sealed class StreamingController
// layer that now owns the landblock.
break;
}
_applyTerrain(loaded.Build, loaded.MeshData);
_state.AddLandblock(
loaded.Landblock,
loaded.Build.EnvCells?.Shells.Select(shell => shell.GeometryId),
loaded.Tier);
EnsureEnvCellMeshesAfterPin(loaded.Build);
// #138: after the landblock is in _loaded (so live projection
// hot-paths), restore any retained server objects ACE won't
// re-send. Fired AFTER AddLandblock, never before.
_onLandblockLoaded?.Invoke(loaded.Landblock.LandblockId);
_presentation.PublishLoaded(loaded);
break;
case LandblockStreamResult.Promoted promoted:
// PromoteToNear carries a complete build and mesh because the
@ -917,28 +982,12 @@ public sealed class StreamingController
// that Far job never started, publish this as the real Near
// landblock; if the base is already resident, merge only the
// Near layer so existing live projections retain identity.
_applyTerrain(promoted.Build, promoted.MeshData);
var promotedRenderIds =
promoted.Build.EnvCells?.Shells.Select(shell => shell.GeometryId);
if (_state.IsLoaded(promoted.LandblockId))
{
_state.AddEntitiesToExistingLandblock(
promoted.LandblockId,
promoted.Entities,
promotedRenderIds);
}
else
{
_state.AddLandblock(
promoted.Landblock,
promotedRenderIds,
LandblockStreamTier.Near);
}
EnsureEnvCellMeshesAfterPin(promoted.Build);
_onLandblockLoaded?.Invoke(promoted.LandblockId);
_presentation.PublishPromoted(
promoted,
mergeIntoExistingLandblock: _state.IsLoaded(promoted.LandblockId));
break;
case LandblockStreamResult.Unloaded unloaded:
_retirements.BeginFull(unloaded.LandblockId);
_presentation.BeginFullRetirement(unloaded.LandblockId);
break;
case LandblockStreamResult.Failed failed:
Console.WriteLine(
@ -951,25 +1000,21 @@ public sealed class StreamingController
}
}
private void ApplyAsFar(LandblockBuild completedBuild, LandblockMeshData meshData)
private void ReconcileCompletedPendingPublication(uint landblockId)
{
var completed = completedBuild.Landblock;
var farLandblock = new LoadedLandblock(
completed.LandblockId,
completed.Heightmap,
Array.Empty<WorldEntity>(),
PhysicsDatBundle.Empty);
var farBuild = new LandblockBuild(farLandblock);
if (_region is null
|| !_region.TryGetDesiredTier(landblockId, out LandblockStreamTier desiredTier))
{
if (_state.IsLoaded(landblockId))
_presentation.BeginFullRetirement(landblockId);
return;
}
_applyTerrain(farBuild, meshData);
_state.AddLandblock(farLandblock, tier: LandblockStreamTier.Far);
_onLandblockLoaded?.Invoke(farLandblock.LandblockId);
}
private void EnsureEnvCellMeshesAfterPin(LandblockBuild build)
{
if (build.EnvCells is { Shells.Length: > 0 } envCells)
_ensureEnvCellMeshes?.Invoke(envCells);
if (desiredTier == LandblockStreamTier.Far
&& _state.IsNearTier(landblockId))
{
_presentation.BeginNearLayerRetirement(landblockId);
}
}
private bool IsStaleGeneration(LandblockStreamResult result) =>
@ -978,7 +1023,7 @@ public sealed class StreamingController
private bool IsPublicationBlockedByRetirement(LandblockStreamResult result) =>
result is LandblockStreamResult.Loaded or LandblockStreamResult.Promoted
&& _retirements.IsPending(result.LandblockId);
&& _presentation.IsRetirementPending(result.LandblockId);
/// <summary>
/// Returns the landblock id associated with <paramref name="result"/>.