621f70eab6
Record route-lifetime streaming overruns and maximum operation costs, require canonical reveal/resource convergence at every connected checkpoint, and keep recenter semantics independent of the production wall-clock budget. Co-authored-by: Erik Nilsson <erikn@users.noreply.github.com>
1968 lines
77 KiB
C#
1968 lines
77 KiB
C#
using System;
|
||
using System.Collections.Generic;
|
||
using System.Diagnostics;
|
||
using AcDream.App.Rendering.Wb;
|
||
using AcDream.Core.Terrain;
|
||
using AcDream.Core.World;
|
||
|
||
namespace AcDream.App.Streaming;
|
||
|
||
/// <summary>
|
||
/// Called once per frame from <c>GameWindow.OnUpdate</c>. Owns the
|
||
/// <see cref="StreamingRegion"/> and uses delegates into
|
||
/// <see cref="LandblockStreamer"/> so tests can inject fakes. All work
|
||
/// happens on the render thread; the streamer itself is background.
|
||
///
|
||
/// <remarks>
|
||
/// Threading: not thread-safe. All calls must happen on the render thread.
|
||
/// </remarks>
|
||
/// </summary>
|
||
public sealed class StreamingController
|
||
: IStreamingFrameBackend,
|
||
IWorldRevealStreamingScheduler
|
||
{
|
||
private readonly record struct DestinationReservation(
|
||
long RevealGeneration,
|
||
uint LandblockId,
|
||
int Radius);
|
||
|
||
private sealed class OriginRecenterRetirement
|
||
{
|
||
public bool RadiiConverged;
|
||
public bool GenerationAdvanced;
|
||
public bool PendingLoadsCleared;
|
||
public bool CompletionQueueCleared;
|
||
public bool RegionCleared;
|
||
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;
|
||
private readonly Action? _clearPendingLoads;
|
||
private readonly LandblockPresentationPipeline _presentation;
|
||
private readonly Func<LandblockStreamResult, bool>
|
||
_isPublicationBlockedByRetirement;
|
||
private readonly StreamingWorkBudgetOptions _configuredWorkBudgetOptions;
|
||
private StreamingWorkBudget _workBudget;
|
||
private StreamingWorkMeter? _activeWorkMeter;
|
||
private StreamingWorkMeterSnapshot _lastWorkMeter;
|
||
private readonly StreamingCompletionQueue _completionQueue = new();
|
||
private long _nextCompletionSequence;
|
||
private int _legacyCompletionProfile = 4;
|
||
private DestinationReservation? _destinationReservation;
|
||
private long _lifetimeWorkOverruns;
|
||
private long _lifetimeOversizedProgress;
|
||
private double _maximumWorkFrameMilliseconds;
|
||
private string? _maximumWorkFrameStage;
|
||
private double _maximumWorkOperationMilliseconds;
|
||
private string? _maximumWorkOperationStage;
|
||
|
||
private readonly GpuWorldState _state;
|
||
private StreamingRegion? _region;
|
||
private RadiiReconfiguration? _pendingRadiiReconfiguration;
|
||
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
|
||
// unload cannot mutate the replacement window after a portal recenter.
|
||
private ulong _generation;
|
||
|
||
// True while streaming is collapsed to the single dungeon landblock the
|
||
// player stands in (the dungeon gate, #133 FPS). AC dungeons have NO
|
||
// adjacent landblocks — neighbors are unrelated ocean-grid dungeons that
|
||
// are never visible, so we stop loading the 25×25 window entirely.
|
||
private bool _collapsed;
|
||
|
||
// The dungeon landblock id we collapsed onto. Once collapsed we key the
|
||
// gate on this STABLE landblock, not the per-frame insideDungeon signal:
|
||
// CurrCell can momentarily resolve to null/outdoor mid-frame, and gating
|
||
// expand on that flicker thrashes collapse↔expand (reload storms + a light
|
||
// leak). We only expand when the observer actually moves to a different
|
||
// landblock (teleport/portal out).
|
||
private uint _collapsedCenter;
|
||
|
||
/// <summary>
|
||
/// Near-tier radius (LBs from observer that load full detail: terrain +
|
||
/// scenery + entities). Runtime changes go through
|
||
/// <see cref="ReconfigureRadii"/>.
|
||
/// </summary>
|
||
public int NearRadius { get; private set; }
|
||
|
||
/// <summary>
|
||
/// Far-tier radius (LBs from observer that load terrain only).
|
||
/// </summary>
|
||
public int FarRadius { get; private set; }
|
||
|
||
/// <summary>
|
||
/// Compatibility quality selector for callers that still express the
|
||
/// streaming profile as a completion count. Four preserves the configured
|
||
/// High profile; other positive values scale every typed time/count/byte
|
||
/// dimension together. It is not an independent execution throttle.
|
||
/// </summary>
|
||
public int MaxCompletionsPerFrame
|
||
{
|
||
get => _legacyCompletionProfile;
|
||
set
|
||
{
|
||
StreamingWorkBudgetOptions profile =
|
||
_configuredWorkBudgetOptions
|
||
.ScaleForLegacyCompletionCount(value);
|
||
_legacyCompletionProfile = value;
|
||
_workBudget = profile.ToBudget();
|
||
}
|
||
}
|
||
|
||
internal long ActiveRevealGeneration =>
|
||
_destinationReservation?.RevealGeneration ?? 0L;
|
||
|
||
internal uint DestinationLandblockId =>
|
||
_destinationReservation?.LandblockId ?? 0u;
|
||
|
||
internal int DestinationRadius =>
|
||
_destinationReservation?.Radius ?? 0;
|
||
|
||
internal void BeginDestinationReservation(
|
||
long revealGeneration,
|
||
uint destinationCell,
|
||
int requiredRenderRadius)
|
||
{
|
||
if (revealGeneration <= 0)
|
||
throw new ArgumentOutOfRangeException(nameof(revealGeneration));
|
||
if (destinationCell == 0u)
|
||
throw new ArgumentOutOfRangeException(nameof(destinationCell));
|
||
if (requiredRenderRadius < 0)
|
||
throw new ArgumentOutOfRangeException(nameof(requiredRenderRadius));
|
||
|
||
_destinationReservation = new DestinationReservation(
|
||
revealGeneration,
|
||
(destinationCell & 0xFFFF0000u) | 0xFFFFu,
|
||
requiredRenderRadius);
|
||
}
|
||
|
||
internal void EndDestinationReservation(long revealGeneration)
|
||
{
|
||
if (_destinationReservation is { } active
|
||
&& active.RevealGeneration == revealGeneration)
|
||
{
|
||
_destinationReservation = null;
|
||
}
|
||
}
|
||
|
||
void IWorldRevealStreamingScheduler.BeginDestinationReservation(
|
||
long revealGeneration,
|
||
uint destinationCell,
|
||
int requiredRenderRadius) =>
|
||
BeginDestinationReservation(
|
||
revealGeneration,
|
||
destinationCell,
|
||
requiredRenderRadius);
|
||
|
||
void IWorldRevealStreamingScheduler.EndDestinationReservation(
|
||
long revealGeneration) =>
|
||
EndDestinationReservation(revealGeneration);
|
||
|
||
// [FRAME-DIAG] (read by GameWindow's ACDREAM_WB_DIAG rollup): the standing
|
||
// accepted completion backlog. A non-zero value during/after a teleport is
|
||
// the typed-budget publication tail. Removable compatibility probe surface.
|
||
public int DeferredApplyBacklog => _completionQueue.Count;
|
||
public int PendingRetirementCount => _presentation.PendingRetirementCount;
|
||
public StreamingWorkDiagnostics WorkDiagnostics
|
||
{
|
||
get
|
||
{
|
||
StreamingCompletionQueueSnapshot queued =
|
||
_completionQueue.CaptureSnapshot();
|
||
return new StreamingWorkDiagnostics(
|
||
_lastWorkMeter,
|
||
_lifetimeWorkOverruns,
|
||
_lifetimeOversizedProgress,
|
||
_maximumWorkFrameMilliseconds,
|
||
_maximumWorkFrameStage,
|
||
_maximumWorkOperationMilliseconds,
|
||
_maximumWorkOperationStage,
|
||
queued.Count,
|
||
queued.RetainedCpuBytes,
|
||
queued.OldestAgeMilliseconds,
|
||
_presentation.PendingPublicationCount,
|
||
_presentation.PendingRetirementCount,
|
||
_completionSource.BacklogCount,
|
||
queued.Destination,
|
||
queued.Control,
|
||
queued.Unload,
|
||
queued.Near,
|
||
queued.Far);
|
||
}
|
||
}
|
||
internal bool IsCollapsedToDungeon => _collapsed;
|
||
|
||
/// <summary>
|
||
/// True once every in-bounds landblock in the requested Chebyshev ring has
|
||
/// crossed the render-thread publication barrier. Worker completion and
|
||
/// world-state registration are not sufficient: all static GfxObj and
|
||
/// EnvCell shell meshes must have completed their render-thread upload.
|
||
/// Portal-space exit uses this alongside physics residency so the world
|
||
/// cannot be revealed while its render slots are still absent.
|
||
/// </summary>
|
||
public bool IsRenderNeighborhoodResident(uint cellOrLandblockId, int radius)
|
||
{
|
||
if (radius < 0)
|
||
throw new ArgumentOutOfRangeException(nameof(radius));
|
||
|
||
// LandDefs::InboundValidCellId validates both map axes and the low-word
|
||
// class (outdoor cell, EnvCell, or canonical landblock sentinel).
|
||
if (!AcDream.Core.Physics.LandDefs.InboundValidCellId(cellOrLandblockId))
|
||
return false;
|
||
int cx = (int)((cellOrLandblockId >> 24) & 0xFFu);
|
||
int cy = (int)((cellOrLandblockId >> 16) & 0xFFu);
|
||
for (int dx = -radius; dx <= radius; dx++)
|
||
for (int dy = -radius; dy <= radius; dy++)
|
||
{
|
||
int nx = cx + dx;
|
||
int ny = cy + dy;
|
||
// Match PhysicsEngine.IsNeighborhoodTerrainResident: the outer
|
||
// 0xFF coordinate has no loadable neighbour beyond it.
|
||
if (nx < 0 || nx > 254 || ny < 0 || ny > 254)
|
||
continue;
|
||
|
||
uint canonical = ((uint)nx << 24) | ((uint)ny << 16) | 0xFFFFu;
|
||
if (!_state.IsNearTier(canonical) || !_state.IsRenderReady(canonical))
|
||
return false;
|
||
}
|
||
|
||
return true;
|
||
}
|
||
|
||
// [FRAME-DIAG]: full-window presentation retirements include explicit
|
||
// reloads and shared-origin recenter barriers. Both retire and later
|
||
// re-upload the complete resident window.
|
||
public int FullWindowRetirementCount { get; private set; }
|
||
public int LastFullWindowRetirementLandblockCount { get; private set; }
|
||
|
||
/// <summary>
|
||
/// Internal compatibility seam for hermetic controller policy tests. Live
|
||
/// composition cannot supply presentation callbacks; it must use the
|
||
/// concrete pipeline constructor below.
|
||
/// </summary>
|
||
internal StreamingController(
|
||
Action<uint, LandblockStreamJobKind, ulong> enqueueLoad,
|
||
Action<uint, ulong> enqueueUnload,
|
||
Func<int, IReadOnlyList<LandblockStreamResult>> drainCompletions,
|
||
Action<LandblockBuild, LandblockMeshData> applyTerrain,
|
||
GpuWorldState state,
|
||
int nearRadius,
|
||
int farRadius,
|
||
Action<uint>? removeTerrain = null,
|
||
Action<uint>? demoteNearLayer = null,
|
||
Action? clearPendingLoads = null,
|
||
Action<uint>? onLandblockLoaded = null,
|
||
Action<EnvCellLandblockBuild>? ensureEnvCellMeshes = null,
|
||
LandblockRetirementCoordinator? retirementCoordinator = null,
|
||
StreamingWorkBudgetOptions? workBudgetOptions = null)
|
||
: this(
|
||
enqueueLoad,
|
||
enqueueUnload,
|
||
drainCompletions,
|
||
state,
|
||
nearRadius,
|
||
farRadius,
|
||
new LandblockPresentationPipeline(
|
||
applyTerrain,
|
||
state,
|
||
onLandblockLoaded,
|
||
ensureEnvCellMeshes,
|
||
retirementCoordinator,
|
||
removeTerrain,
|
||
demoteNearLayer),
|
||
clearPendingLoads,
|
||
workBudgetOptions)
|
||
{
|
||
}
|
||
|
||
/// <summary>
|
||
/// Concrete presentation path. Its pipeline is the sole presentation and
|
||
/// retirement owner, so no legacy publication delegate can be supplied or
|
||
/// silently ignored.
|
||
/// </summary>
|
||
internal StreamingController(
|
||
Action<uint, LandblockStreamJobKind, ulong> enqueueLoad,
|
||
Action<uint, ulong> enqueueUnload,
|
||
Func<int, IReadOnlyList<LandblockStreamResult>> drainCompletions,
|
||
GpuWorldState state,
|
||
int nearRadius,
|
||
int farRadius,
|
||
LandblockPresentationPipeline presentationPipeline,
|
||
Action? clearPendingLoads = null,
|
||
StreamingWorkBudgetOptions? workBudgetOptions = null)
|
||
: this(
|
||
enqueueLoad,
|
||
enqueueUnload,
|
||
new DelegateLandblockCompletionSource(drainCompletions),
|
||
state,
|
||
nearRadius,
|
||
farRadius,
|
||
presentationPipeline,
|
||
clearPendingLoads,
|
||
workBudgetOptions)
|
||
{
|
||
}
|
||
|
||
/// <summary>
|
||
/// Allocation-free production completion path. Peek/read is single-consumer
|
||
/// and lets the scheduler price a result before adopting it.
|
||
/// </summary>
|
||
public StreamingController(
|
||
Action<uint, LandblockStreamJobKind, ulong> enqueueLoad,
|
||
Action<uint, ulong> enqueueUnload,
|
||
ILandblockCompletionSource completionSource,
|
||
GpuWorldState state,
|
||
int nearRadius,
|
||
int farRadius,
|
||
LandblockPresentationPipeline presentationPipeline,
|
||
Action? clearPendingLoads = null,
|
||
StreamingWorkBudgetOptions? workBudgetOptions = null)
|
||
{
|
||
_enqueueLoad = enqueueLoad;
|
||
_enqueueUnload = enqueueUnload;
|
||
_completionSource = completionSource
|
||
?? throw new ArgumentNullException(nameof(completionSource));
|
||
_clearPendingLoads = clearPendingLoads;
|
||
_state = state;
|
||
_presentation = presentationPipeline
|
||
?? throw new ArgumentNullException(nameof(presentationPipeline));
|
||
_isPublicationBlockedByRetirement =
|
||
IsPublicationBlockedByRetirement;
|
||
_configuredWorkBudgetOptions =
|
||
workBudgetOptions ?? StreamingWorkBudgetOptions.Default;
|
||
_workBudget = _configuredWorkBudgetOptions.ToBudget();
|
||
if (!_presentation.MatchesState(_state))
|
||
{
|
||
throw new ArgumentException(
|
||
"The presentation pipeline must own the controller's world state.",
|
||
nameof(presentationPipeline));
|
||
}
|
||
NearRadius = nearRadius;
|
||
FarRadius = farRadius;
|
||
}
|
||
|
||
/// <summary>
|
||
/// Reconciles an active outdoor streaming window to new quality radii
|
||
/// without replacing the worker/controller or blindly bootstrapping data
|
||
/// already published in <see cref="GpuWorldState"/>. Pending jobs from the
|
||
/// old window are invalidated by generation; resident blocks are retained,
|
||
/// promoted, demoted, loaded, or unloaded from their actual current tier.
|
||
/// A sealed dungeon remains radius-zero until its normal exit edge while
|
||
/// remembering the new outdoor radii for that exit.
|
||
/// </summary>
|
||
public void ReconfigureRadii(int nearRadius, int farRadius)
|
||
{
|
||
if (nearRadius < 0)
|
||
throw new ArgumentOutOfRangeException(nameof(nearRadius));
|
||
if (farRadius < nearRadius)
|
||
throw new ArgumentOutOfRangeException(
|
||
nameof(farRadius),
|
||
"Far radius must be greater than or equal to near radius.");
|
||
|
||
// A radius transaction may not admit work after an origin-recenter
|
||
// snapshot has been captured. Retain only the latest requested radii;
|
||
// the first Tick after origin commit applies them to the new frame.
|
||
if (_originRecenterRetirement is not null)
|
||
{
|
||
_deferredRadiiRequest = (nearRadius, farRadius);
|
||
return;
|
||
}
|
||
|
||
// Queue callbacks are injected seams and can synchronously reenter the
|
||
// controller. Last-request-wins deferral keeps the active mutation
|
||
// cursor stable; the request is applied after the current transaction
|
||
// converges instead of recursively replaying its active step.
|
||
if (_advancingRadiiReconfiguration)
|
||
{
|
||
_deferredRadiiRequest = (nearRadius, farRadius);
|
||
return;
|
||
}
|
||
|
||
// A prior callback may have failed after this controller accepted the
|
||
// request. Resume that exact transaction before considering another
|
||
// target; replacing it would orphan already-admitted generation work.
|
||
if (_pendingRadiiReconfiguration is not null)
|
||
AdvanceRadiiReconfiguration();
|
||
|
||
// This explicit call is newer than any request deferred by a prior
|
||
// failed attempt and therefore supersedes it.
|
||
_deferredRadiiRequest = null;
|
||
|
||
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.
|
||
if (!ConvergePendingPublications())
|
||
{
|
||
_deferredRadiiRequest = (nearRadius, farRadius);
|
||
return;
|
||
}
|
||
|
||
if (_collapsed || _region is null)
|
||
{
|
||
NearRadius = nearRadius;
|
||
FarRadius = farRadius;
|
||
return;
|
||
}
|
||
|
||
var rebuilt = new StreamingRegion(
|
||
_region.CenterX,
|
||
_region.CenterY,
|
||
nearRadius,
|
||
farRadius);
|
||
TwoTierDiff bootstrap = rebuilt.ComputeFirstTickDiff();
|
||
var desiredNear = new HashSet<uint>(bootstrap.ToLoadNear);
|
||
var desiredFar = new HashSet<uint>(bootstrap.ToLoadFar);
|
||
var mutations = new List<RadiiMutation>();
|
||
|
||
uint[] loaded = [.. _state.LoadedLandblockIds];
|
||
for (int i = 0; i < loaded.Length; i++)
|
||
{
|
||
uint id = loaded[i];
|
||
if (desiredNear.Contains(id))
|
||
{
|
||
if (!_state.IsNearTier(id))
|
||
mutations.Add(new RadiiMutation(
|
||
() => EnqueueLoad(id, LandblockStreamJobKind.PromoteToNear)));
|
||
}
|
||
else if (desiredFar.Contains(id))
|
||
{
|
||
if (_state.IsNearTier(id))
|
||
mutations.Add(new RadiiMutation(() => DemoteLandblock(id)));
|
||
}
|
||
else
|
||
{
|
||
mutations.Add(new RadiiMutation(() => EnqueueUnload(id)));
|
||
}
|
||
}
|
||
|
||
foreach (uint id in desiredNear)
|
||
if (!_state.IsLoaded(id))
|
||
mutations.Add(new RadiiMutation(
|
||
() => EnqueueLoad(id, LandblockStreamJobKind.LoadNear)));
|
||
foreach (uint id in desiredFar)
|
||
if (!_state.IsLoaded(id))
|
||
mutations.Add(new RadiiMutation(
|
||
() => EnqueueLoad(id, LandblockStreamJobKind.LoadFar)));
|
||
|
||
_pendingRadiiReconfiguration = new RadiiReconfiguration(
|
||
nearRadius,
|
||
farRadius,
|
||
rebuilt,
|
||
mutations);
|
||
AdvanceRadiiReconfiguration();
|
||
}
|
||
|
||
/// <summary>
|
||
/// Compatibility constructor for deterministic tests and callers that do
|
||
/// not own an asynchronous worker. Production must use the generation-aware
|
||
/// overload above so hard-recenter results can be rejected by incarnation.
|
||
/// </summary>
|
||
internal StreamingController(
|
||
Action<uint, LandblockStreamJobKind> enqueueLoad,
|
||
Action<uint> enqueueUnload,
|
||
Func<int, IReadOnlyList<LandblockStreamResult>> drainCompletions,
|
||
Action<LandblockBuild, LandblockMeshData> applyTerrain,
|
||
GpuWorldState state,
|
||
int nearRadius,
|
||
int farRadius,
|
||
Action<uint>? removeTerrain = null,
|
||
Action<uint>? demoteNearLayer = null,
|
||
Action? clearPendingLoads = null,
|
||
Action<uint>? onLandblockLoaded = null,
|
||
Action<EnvCellLandblockBuild>? ensureEnvCellMeshes = null,
|
||
LandblockRetirementCoordinator? retirementCoordinator = null,
|
||
StreamingWorkBudgetOptions? workBudgetOptions = null)
|
||
: this(
|
||
(id, kind, _) => enqueueLoad(id, kind),
|
||
(id, _) => enqueueUnload(id),
|
||
drainCompletions,
|
||
applyTerrain,
|
||
state,
|
||
nearRadius,
|
||
farRadius,
|
||
removeTerrain,
|
||
demoteNearLayer,
|
||
clearPendingLoads,
|
||
onLandblockLoaded,
|
||
ensureEnvCellMeshes,
|
||
retirementCoordinator,
|
||
workBudgetOptions)
|
||
{
|
||
}
|
||
|
||
private void EnqueueLoad(uint id, LandblockStreamJobKind kind) =>
|
||
_enqueueLoad(id, kind, _generation);
|
||
|
||
private void EnqueueUnload(uint id) => _enqueueUnload(id, _generation);
|
||
|
||
private void DemoteLandblock(uint id)
|
||
{
|
||
uint canonical = (id & 0xFFFF0000u) | 0xFFFFu;
|
||
_presentation.EnqueueNearLayerRetirement(canonical);
|
||
}
|
||
|
||
private bool AdvanceGeneration()
|
||
{
|
||
if (!ConvergePendingPublications())
|
||
return false;
|
||
_generation = unchecked(_generation + 1);
|
||
return true;
|
||
}
|
||
|
||
private bool ConvergePendingPublications()
|
||
{
|
||
IReadOnlyList<LandblockStreamResult> pending =
|
||
_presentation.GetPendingPublicationResults();
|
||
if (pending.Count == 0)
|
||
return true;
|
||
if (_activeWorkMeter is null)
|
||
{
|
||
// Settings/native callbacks may request policy changes between
|
||
// frames, but publication remains owned by the frame-scoped meter.
|
||
return false;
|
||
}
|
||
|
||
try
|
||
{
|
||
bool progressed = false;
|
||
for (int i = 0; i < pending.Count; i++)
|
||
{
|
||
LandblockStreamResult result = pending[i];
|
||
using StreamingWorkMeter.LaneScope lane =
|
||
_activeWorkMeter.EnterLane(
|
||
IsDestinationWork(result.LandblockId)
|
||
? StreamingWorkLane.Destination
|
||
: StreamingWorkLane.NonDestination);
|
||
LandblockPublicationAdvance advance =
|
||
_presentation.ResumePublication(
|
||
result,
|
||
_activeWorkMeter,
|
||
ensureProgress: !progressed);
|
||
progressed |= advance.Progressed;
|
||
if (!advance.Completed)
|
||
return false;
|
||
}
|
||
return true;
|
||
}
|
||
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.
|
||
_completionQueue.RemoveResults(
|
||
pending,
|
||
result => !_presentation.HasPendingPublication(result));
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Advance one frame. <paramref name="observerCx"/>/<paramref name="observerCy"/>
|
||
/// are landblock coordinates (0..255) of the current viewer — the camera
|
||
/// in offline mode, the server-sent player position in live.
|
||
///
|
||
/// <para>Two-tier model (Phase A.5 T13):</para>
|
||
/// <list type="bullet">
|
||
/// <item><see cref="TwoTierDiff.ToLoadFar"/> → enqueue LoadFar (terrain only, no entities)</item>
|
||
/// <item><see cref="TwoTierDiff.ToLoadNear"/> → enqueue LoadNear (terrain + entities)</item>
|
||
/// <item><see cref="TwoTierDiff.ToPromote"/> → enqueue PromoteToNear (entity layer for already-loaded terrain)</item>
|
||
/// <item><see cref="TwoTierDiff.ToDemote"/> → drop entities on render thread immediately (terrain stays)</item>
|
||
/// <item><see cref="TwoTierDiff.ToUnload"/> → enqueue full unload</item>
|
||
/// </list>
|
||
/// </summary>
|
||
public void Tick(int observerCx, int observerCy, bool insideDungeon = false)
|
||
{
|
||
if (_activeWorkMeter is not null)
|
||
throw new InvalidOperationException(
|
||
"StreamingController.Tick cannot be reentered.");
|
||
|
||
var meter = new StreamingWorkMeter(
|
||
_workBudget,
|
||
destinationReservationActive: _destinationReservation is not null);
|
||
_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)
|
||
{
|
||
bool retirementWasPending =
|
||
_presentation.PendingRetirementCount != 0;
|
||
TryAdvanceOriginRecenterPreparation(meter);
|
||
if (_presentation.UsesBudgetedRetirementSteps
|
||
|| retirementWasPending)
|
||
{
|
||
_presentation.AdvanceRetirements(meter);
|
||
}
|
||
return;
|
||
}
|
||
|
||
if (_pendingRadiiReconfiguration is not null)
|
||
AdvanceRadiiReconfiguration();
|
||
else if (_deferredRadiiRequest is { } deferred)
|
||
{
|
||
_deferredRadiiRequest = null;
|
||
ReconfigureRadii(deferred.NearRadius, deferred.FarRadius);
|
||
}
|
||
|
||
// 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;
|
||
|
||
uint centerId = StreamingRegion.EncodeLandblockId(observerCx, observerCy);
|
||
|
||
if (_collapsed)
|
||
{
|
||
// Hysteresis. Cases:
|
||
// - Still in the SAME dungeon landblock → hold (sweep stragglers).
|
||
// - In a DIFFERENT dungeon cell (multi-landblock dungeon / new dungeon)
|
||
// → re-collapse onto it.
|
||
// - CurrCell flickered null but the player hasn't gone anywhere: the
|
||
// observer landblock reverts to the position-derived value, which for a
|
||
// dungeon is only ever the ADJACENT off-by-one landblock (negative cell-
|
||
// local Y). Hold — never expand on an adjacent flicker.
|
||
// - Genuinely left to a DISTANT landblock (portal/teleport out, always far
|
||
// from the ocean-grid dungeon block) → expand.
|
||
if (insideDungeon && centerId != _collapsedCenter)
|
||
EnterDungeonCollapse(observerCx, observerCy, centerId);
|
||
else if (!insideDungeon && ChebyshevLandblocks(centerId, _collapsedCenter) > 1)
|
||
ExitDungeonExpand(observerCx, observerCy);
|
||
else
|
||
SweepCollapsed();
|
||
}
|
||
else if (insideDungeon)
|
||
{
|
||
EnterDungeonCollapse(observerCx, observerCy, centerId);
|
||
}
|
||
else
|
||
{
|
||
NormalTick(observerCx, observerCy);
|
||
}
|
||
|
||
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
|
||
{
|
||
meter.FinishFrame();
|
||
StreamingWorkMeterSnapshot snapshot = meter.Snapshot;
|
||
ObserveWorkLifetime(snapshot);
|
||
_lastWorkMeter = snapshot;
|
||
_activeWorkMeter = null;
|
||
}
|
||
}
|
||
|
||
private void ObserveWorkLifetime(StreamingWorkMeterSnapshot snapshot)
|
||
{
|
||
_lifetimeWorkOverruns = SaturatingAdd(
|
||
_lifetimeWorkOverruns,
|
||
snapshot.OverrunCount);
|
||
_lifetimeOversizedProgress = SaturatingAdd(
|
||
_lifetimeOversizedProgress,
|
||
snapshot.OversizedProgressCount);
|
||
|
||
if (_maximumWorkFrameStage is null
|
||
|| snapshot.ElapsedMilliseconds > _maximumWorkFrameMilliseconds)
|
||
{
|
||
_maximumWorkFrameMilliseconds = snapshot.ElapsedMilliseconds;
|
||
_maximumWorkFrameStage = snapshot.LastStage;
|
||
}
|
||
|
||
if (_maximumWorkOperationStage is null
|
||
|| snapshot.MaximumOperationMilliseconds
|
||
> _maximumWorkOperationMilliseconds)
|
||
{
|
||
_maximumWorkOperationMilliseconds =
|
||
snapshot.MaximumOperationMilliseconds;
|
||
_maximumWorkOperationStage = snapshot.MaximumOperationStage;
|
||
}
|
||
}
|
||
|
||
private static long SaturatingAdd(long left, int right) =>
|
||
left > long.MaxValue - right ? long.MaxValue : left + right;
|
||
|
||
private void AdvanceRadiiReconfiguration()
|
||
{
|
||
if (_advancingRadiiReconfiguration)
|
||
return;
|
||
|
||
RadiiReconfiguration pending = _pendingRadiiReconfiguration
|
||
?? throw new InvalidOperationException("No radii reconfiguration is pending.");
|
||
var failures = new List<Exception>();
|
||
_advancingRadiiReconfiguration = true;
|
||
try
|
||
{
|
||
|
||
if (!pending.GenerationAdvanced)
|
||
{
|
||
if (!AdvanceGeneration())
|
||
return;
|
||
pending.GenerationAdvanced = true;
|
||
}
|
||
|
||
if (!pending.PendingLoadsCleared)
|
||
{
|
||
try
|
||
{
|
||
_clearPendingLoads?.Invoke();
|
||
pending.PendingLoadsCleared = true;
|
||
}
|
||
catch (Exception error)
|
||
{
|
||
if (error is StreamingMutationException { MutationCommitted: true })
|
||
pending.PendingLoadsCleared = true;
|
||
failures.Add(error);
|
||
}
|
||
}
|
||
|
||
// Do not admit new-generation work until cancellation of the old
|
||
// inbox is durable. Otherwise a successful retry of ClearPendingLoads
|
||
// would also erase mutations already marked complete by this ledger.
|
||
for (int i = 0; pending.PendingLoadsCleared && i < pending.Mutations.Count; i++)
|
||
{
|
||
RadiiMutation mutation = pending.Mutations[i];
|
||
if (mutation.Completed)
|
||
continue;
|
||
try
|
||
{
|
||
mutation.Apply();
|
||
mutation.Completed = true;
|
||
}
|
||
catch (Exception error)
|
||
{
|
||
if (error is StreamingMutationException { MutationCommitted: true })
|
||
mutation.Completed = true;
|
||
failures.Add(error);
|
||
}
|
||
}
|
||
|
||
bool converged = pending.PendingLoadsCleared
|
||
&& pending.Mutations.All(static mutation => mutation.Completed);
|
||
if (converged)
|
||
{
|
||
// Publish the new logical window only after every queue/retirement
|
||
// admission is durable. A later Tick can now trust Resident as the
|
||
// complete desired set and will never strand a hole.
|
||
pending.Region.MarkResidentFromBootstrap();
|
||
_region = pending.Region;
|
||
NearRadius = pending.NearRadius;
|
||
FarRadius = pending.FarRadius;
|
||
_completionQueue.Clear();
|
||
_pendingRadiiReconfiguration = null;
|
||
}
|
||
}
|
||
finally
|
||
{
|
||
_advancingRadiiReconfiguration = false;
|
||
}
|
||
|
||
if (failures.Count != 0)
|
||
{
|
||
throw new AggregateException(
|
||
"Streaming quality reconfiguration did not complete cleanly.",
|
||
failures);
|
||
}
|
||
|
||
if (_pendingRadiiReconfiguration is null
|
||
&& _deferredRadiiRequest is { } deferred)
|
||
{
|
||
_deferredRadiiRequest = null;
|
||
ReconfigureRadii(deferred.NearRadius, deferred.FarRadius);
|
||
}
|
||
}
|
||
|
||
private sealed class RadiiReconfiguration
|
||
{
|
||
public RadiiReconfiguration(
|
||
int nearRadius,
|
||
int farRadius,
|
||
StreamingRegion region,
|
||
List<RadiiMutation> mutations)
|
||
{
|
||
NearRadius = nearRadius;
|
||
FarRadius = farRadius;
|
||
Region = region;
|
||
Mutations = mutations;
|
||
}
|
||
|
||
public int NearRadius { get; }
|
||
public int FarRadius { get; }
|
||
public StreamingRegion Region { get; }
|
||
public List<RadiiMutation> Mutations { get; }
|
||
public bool GenerationAdvanced { get; set; }
|
||
public bool PendingLoadsCleared { get; set; }
|
||
}
|
||
|
||
private sealed class RadiiMutation
|
||
{
|
||
public RadiiMutation(Action apply) => Apply = apply;
|
||
|
||
public Action Apply { get; }
|
||
public bool Completed { get; set; }
|
||
}
|
||
|
||
/// <summary>
|
||
/// #135: collapse to a single dungeon landblock IMMEDIATELY, before the first
|
||
/// <see cref="Tick"/> has a chance to bootstrap the full 25×25 window. Called
|
||
/// from the login / teleport spawn path the instant the streaming center is
|
||
/// recentered onto a SEALED dungeon landblock.
|
||
///
|
||
/// <para>The per-frame <c>insideDungeon</c> gate keys on the physics
|
||
/// <c>CurrCell</c>, which is only set once the player is PLACED — and placement
|
||
/// waits for the dungeon landblock to hydrate. So for the whole hydration window
|
||
/// (tens of seconds for a ~200-cell dungeon) the gate reads false and
|
||
/// <see cref="NormalTick"/> would enqueue the ~24 unrelated ocean-grid neighbor
|
||
/// dungeons (+ ~19k entities each); the collapse then only mops them up after
|
||
/// placement. That mop-up is the 10→high FPS ramp users see at a dungeon login.</para>
|
||
///
|
||
/// <para>Pre-collapsing means the EXPENSIVE dungeon-neighbour window is never
|
||
/// enqueued. On teleport nothing is enqueued at all (this fires before the next
|
||
/// Tick recenters). On login a brief Holtburg outdoor window may be enqueued by the
|
||
/// frame-1 NormalTick (before the player's spawn arrives) and is immediately
|
||
/// cancelled by <c>_clearPendingLoads</c> here — cheap outdoor terrain, not the
|
||
/// ocean-grid dungeons, and a handful of already-dequeued loads get swept next
|
||
/// frame. Idempotent: a no-op when already collapsed onto this same landblock, so a
|
||
/// re-sent spawn or a same-frame double call costs nothing. Render-thread only,
|
||
/// same as <see cref="Tick"/>.</para>
|
||
/// </summary>
|
||
public void InitializeKnownLoginCenter(int cx, int cy, bool isSealedDungeon)
|
||
{
|
||
// StreamingReadinessGate keeps the worker stopped until the player's
|
||
// real spawn supplies this center. There is therefore no guessed-center
|
||
// window to force-reload here. A sealed dungeon is the one special case:
|
||
// pin radius zero before the first Tick so no ocean-grid neighbours are
|
||
// ever enqueued.
|
||
if (isSealedDungeon)
|
||
PreCollapseToDungeon(cx, cy);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Pins streaming to a sealed dungeon before the first normal Tick.
|
||
/// </summary>
|
||
public void PreCollapseToDungeon(int cx, int cy)
|
||
{
|
||
uint centerId = StreamingRegion.EncodeLandblockId(cx, cy);
|
||
if (_collapsed && _collapsedCenter == centerId) return;
|
||
EnterDungeonCollapse(cx, cy, centerId);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Outdoor / building-interior streaming — the original two-tier model.
|
||
/// </summary>
|
||
private void NormalTick(int observerCx, int observerCy)
|
||
{
|
||
if (_region is null)
|
||
{
|
||
_region = new StreamingRegion(observerCx, observerCy, NearRadius, FarRadius);
|
||
var bootstrap = _region.ComputeFirstTickDiff();
|
||
foreach (var id in bootstrap.ToLoadNear) EnqueueLoad(id, LandblockStreamJobKind.LoadNear);
|
||
foreach (var id in bootstrap.ToLoadFar) EnqueueLoad(id, LandblockStreamJobKind.LoadFar);
|
||
_region.MarkResidentFromBootstrap();
|
||
}
|
||
else if (_region.CenterX != observerCx || _region.CenterY != observerCy)
|
||
{
|
||
var diff = _region.RecenterTo(observerCx, observerCy);
|
||
foreach (var id in diff.ToPromote) EnqueueLoad(id, LandblockStreamJobKind.PromoteToNear);
|
||
foreach (var id in diff.ToLoadNear) EnqueueLoad(id, LandblockStreamJobKind.LoadNear);
|
||
foreach (var id in diff.ToLoadFar) EnqueueLoad(id, LandblockStreamJobKind.LoadFar);
|
||
foreach (var id in diff.ToDemote) DemoteLandblock(id);
|
||
foreach (var id in diff.ToUnload) EnqueueUnload(id);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Dungeon-entry edge: cancel the in-flight window load, unload every
|
||
/// resident neighbor, and pin streaming to the player's single dungeon
|
||
/// landblock. Retail-faithful — AC dungeons have no adjacent landblocks
|
||
/// (ACE <c>LandblockManager.GetAdjacentIDs</c> returns empty for a dungeon);
|
||
/// the 25×25 window was pulling in ~129 unrelated ocean-grid dungeons and
|
||
/// their thousands of emitters (#133 FPS). Unloading them also tears down
|
||
/// their lights, shrinking the static-light set toward retail's ≤40.
|
||
/// </summary>
|
||
private void EnterDungeonCollapse(int cx, int cy, uint centerId)
|
||
{
|
||
bool logTransition = !_collapsed || _collapsedCenter != centerId;
|
||
if (logTransition)
|
||
Console.WriteLine($"streaming: dungeon collapse -> 0x{centerId:X8}");
|
||
if (!AdvanceGeneration())
|
||
return;
|
||
_collapsed = true;
|
||
_collapsedCenter = centerId;
|
||
_clearPendingLoads?.Invoke();
|
||
_completionQueue.Clear();
|
||
_region = null;
|
||
|
||
foreach (var id in _state.LoadedLandblockIds)
|
||
if (id != centerId) EnqueueUnload(id);
|
||
|
||
// Pin a radius-0 region so RecenterTo never re-expands while inside,
|
||
// and so the post-exit rebuild starts from a clean, consistent state.
|
||
_region = new StreamingRegion(cx, cy, 0, 0);
|
||
_region.MarkResidentFromBootstrap();
|
||
|
||
// The dungeon landblock itself must be (or become) loaded. If a prior
|
||
// ClearPendingLoads cancelled its queued load, re-enqueue it.
|
||
if (!_state.IsLoaded(centerId))
|
||
EnqueueLoad(centerId, LandblockStreamJobKind.LoadNear);
|
||
else if (!_state.IsNearTier(centerId))
|
||
EnqueueLoad(centerId, LandblockStreamJobKind.PromoteToNear);
|
||
}
|
||
|
||
/// <summary>
|
||
/// While collapsed, unload any landblock that finished loading after the
|
||
/// collapse edge — a Load the worker had already dequeued before the
|
||
/// <see cref="LandblockStreamer.ClearPendingLoads"/> control job took
|
||
/// effect. At steady state only the dungeon landblock is resident, so this
|
||
/// is a no-op.
|
||
/// </summary>
|
||
private void SweepCollapsed()
|
||
{
|
||
// Always preserve the true dungeon landblock (_collapsedCenter), never the
|
||
// per-frame observer landblock — a CurrCell flicker must not unload the dungeon.
|
||
foreach (var id in _state.LoadedLandblockIds)
|
||
if (id != _collapsedCenter) EnqueueUnload(id);
|
||
}
|
||
|
||
/// <summary>Chebyshev distance in landblock cells between two landblock ids.</summary>
|
||
private static int ChebyshevLandblocks(uint a, uint b)
|
||
{
|
||
int ax = (int)((a >> 24) & 0xFFu), ay = (int)((a >> 16) & 0xFFu);
|
||
int bx = (int)((b >> 24) & 0xFFu), by = (int)((b >> 16) & 0xFFu);
|
||
return Math.Max(Math.Abs(ax - bx), Math.Abs(ay - by));
|
||
}
|
||
|
||
/// <summary>
|
||
/// True when <paramref name="id"/> belongs to the canonical reveal
|
||
/// generation's destination neighborhood.
|
||
/// </summary>
|
||
private bool IsDestinationWork(uint id)
|
||
=> _destinationReservation is { } reservation
|
||
&& ChebyshevLandblocks(id, reservation.LandblockId)
|
||
<= reservation.Radius;
|
||
|
||
/// <summary>
|
||
/// Dungeon-exit edge (portal to outdoors / teleport): rebuild the full
|
||
/// two-tier window at the new center and unload anything resident from the
|
||
/// collapsed state that falls outside it.
|
||
/// </summary>
|
||
private void ExitDungeonExpand(int observerCx, int observerCy)
|
||
{
|
||
Console.WriteLine(
|
||
$"streaming: dungeon EXIT-expand -> ({observerCx},{observerCy}) " +
|
||
$"(was collapsed on 0x{_collapsedCenter:X8})");
|
||
if (!AdvanceGeneration())
|
||
return;
|
||
_collapsed = false;
|
||
var rebuilt = new StreamingRegion(observerCx, observerCy, NearRadius, FarRadius);
|
||
|
||
foreach (var id in _state.LoadedLandblockIds)
|
||
if (!rebuilt.Resident.Contains(id)) EnqueueUnload(id);
|
||
|
||
var boot = rebuilt.ComputeFirstTickDiff();
|
||
foreach (var id in boot.ToLoadNear)
|
||
if (!_state.IsLoaded(id)) EnqueueLoad(id, LandblockStreamJobKind.LoadNear);
|
||
foreach (var id in boot.ToLoadFar)
|
||
if (!_state.IsLoaded(id)) EnqueueLoad(id, LandblockStreamJobKind.LoadFar);
|
||
rebuilt.MarkResidentFromBootstrap();
|
||
_region = rebuilt;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 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>
|
||
public void ForceReloadWindow()
|
||
{
|
||
BeginFullWindowRetirement();
|
||
}
|
||
|
||
/// <summary>
|
||
/// Starts the old-window half of a shared-origin recenter. All accepted
|
||
/// worker work is invalidated and every resident landblock is detached
|
||
/// through the retryable presentation ledger while the old coordinate
|
||
/// frame is still active. <see cref="Tick"/> remains blocked until
|
||
/// <see cref="TryCommitOriginRecenter"/> succeeds.
|
||
/// </summary>
|
||
internal void BeginOriginRecenter()
|
||
{
|
||
_originRecenterRetirement ??= new OriginRecenterRetirement();
|
||
}
|
||
|
||
/// <summary>
|
||
/// Advances every retained old-window teardown and reports whether the
|
||
/// composition root may safely change the shared world origin.
|
||
/// </summary>
|
||
internal bool IsOriginRecenterRetirementComplete()
|
||
{
|
||
if (_originRecenterRetirement is null)
|
||
throw new InvalidOperationException(
|
||
"No streaming-origin recenter transaction is pending.");
|
||
|
||
return _originRecenterRetirement.PreparationCommitted
|
||
&& _presentation.PendingRetirementCount == 0;
|
||
}
|
||
|
||
/// <summary>
|
||
/// Releases the streaming bootstrap gate after the composition root has
|
||
/// committed the new shared origin. No old-window presentation owner may
|
||
/// still be pending at this edge.
|
||
/// </summary>
|
||
internal bool TryCommitOriginRecenter(
|
||
int destinationX,
|
||
int destinationY,
|
||
bool isSealedDungeon)
|
||
{
|
||
if (_advancingOriginRecenter)
|
||
return false;
|
||
|
||
_advancingOriginRecenter = true;
|
||
try
|
||
{
|
||
return TryCommitOriginRecenterCore(
|
||
destinationX,
|
||
destinationY,
|
||
isSealedDungeon);
|
||
}
|
||
finally
|
||
{
|
||
_advancingOriginRecenter = false;
|
||
}
|
||
}
|
||
|
||
private bool TryCommitOriginRecenterCore(
|
||
int destinationX,
|
||
int destinationY,
|
||
bool isSealedDungeon)
|
||
{
|
||
OriginRecenterRetirement transaction = _originRecenterRetirement
|
||
?? throw new InvalidOperationException(
|
||
"No streaming-origin recenter transaction is pending.");
|
||
if (!transaction.PreparationCommitted)
|
||
throw new InvalidOperationException(
|
||
"Streaming-origin retirement preparation has not completed.");
|
||
if (_presentation.PendingRetirementCount != 0)
|
||
throw new InvalidOperationException(
|
||
"The streaming origin cannot change while old-window presentation retirement is pending.");
|
||
|
||
var destination = (destinationX, destinationY, isSealedDungeon);
|
||
if (transaction.Destination is { } retained && retained != destination)
|
||
{
|
||
throw new InvalidOperationException(
|
||
"A recenter transaction cannot commit two different destinations.");
|
||
}
|
||
transaction.Destination ??= destination;
|
||
|
||
if (!transaction.DestinationConfigured)
|
||
{
|
||
_collapsed = isSealedDungeon;
|
||
_collapsedCenter = isSealedDungeon
|
||
? StreamingRegion.EncodeLandblockId(destinationX, destinationY)
|
||
: 0u;
|
||
if (isSealedDungeon)
|
||
{
|
||
_region = new StreamingRegion(
|
||
destinationX,
|
||
destinationY,
|
||
nearRadius: 0,
|
||
farRadius: 0);
|
||
_region.MarkResidentFromBootstrap();
|
||
}
|
||
else
|
||
{
|
||
_region = null;
|
||
}
|
||
transaction.DestinationConfigured = true;
|
||
}
|
||
|
||
if (isSealedDungeon && !transaction.DestinationLoadEnqueued)
|
||
{
|
||
try
|
||
{
|
||
EnqueueLoad(
|
||
StreamingRegion.EncodeLandblockId(destinationX, destinationY),
|
||
LandblockStreamJobKind.LoadNear);
|
||
transaction.DestinationLoadEnqueued = true;
|
||
}
|
||
catch (StreamingMutationException error) when (error.MutationCommitted)
|
||
{
|
||
transaction.DestinationLoadEnqueued = true;
|
||
Console.WriteLine(
|
||
$"streaming: committed dungeon recenter enqueue reported failure: {error}");
|
||
return false;
|
||
}
|
||
catch (Exception error)
|
||
{
|
||
Console.WriteLine(
|
||
$"streaming: dungeon recenter enqueue will resume: {error}");
|
||
return false;
|
||
}
|
||
}
|
||
|
||
_originRecenterRetirement = null;
|
||
return true;
|
||
}
|
||
|
||
/// <summary>
|
||
/// Releases a fully retired origin transaction at a session boundary
|
||
/// without bootstrapping a destination from the ending session.
|
||
/// </summary>
|
||
internal bool TryCancelOriginRecenter()
|
||
{
|
||
if (_originRecenterRetirement is not { PreparationCommitted: true })
|
||
return false;
|
||
if (_presentation.PendingRetirementCount != 0)
|
||
return false;
|
||
|
||
_collapsed = false;
|
||
_collapsedCenter = 0u;
|
||
_region = null;
|
||
_originRecenterRetirement = null;
|
||
return true;
|
||
}
|
||
|
||
private bool TryAdvanceOriginRecenterPreparation(StreamingWorkMeter meter)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(meter);
|
||
OriginRecenterRetirement transaction = _originRecenterRetirement
|
||
?? throw new InvalidOperationException(
|
||
"No streaming-origin recenter transaction is pending.");
|
||
if (transaction.PreparationCommitted)
|
||
return true;
|
||
if (_advancingOriginRecenter)
|
||
return false;
|
||
|
||
_advancingOriginRecenter = true;
|
||
try
|
||
{
|
||
if (!transaction.RadiiConverged)
|
||
{
|
||
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)
|
||
{
|
||
bool ClearPendingLoads()
|
||
{
|
||
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;
|
||
}
|
||
}
|
||
|
||
if (!TryRunStreamingWork(
|
||
meter,
|
||
new StreamingWorkCost(EntityOperations: 1),
|
||
"recenter-clear-worker-inbox",
|
||
ClearPendingLoads))
|
||
return false;
|
||
}
|
||
if (!transaction.CompletionQueueCleared)
|
||
{
|
||
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)
|
||
{
|
||
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 = [];
|
||
transaction.ResidentEnumerator =
|
||
_state.LoadedLandblockIds.GetEnumerator();
|
||
}
|
||
while (transaction.ResidentEnumerator is { } residentEnumerator)
|
||
{
|
||
StreamingWorkAdmission admission = meter.TryReserve(
|
||
new StreamingWorkCost(EntityOperations: 1),
|
||
"recenter-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(
|
||
"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)
|
||
{
|
||
// A delivered visibility-observer failure may surface after
|
||
// detachment has committed. Advance that exact cursor only
|
||
// when world state proves the old resident is unreachable;
|
||
// 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}");
|
||
return false;
|
||
}
|
||
}
|
||
|
||
transaction.PreparationCommitted = true;
|
||
return true;
|
||
}
|
||
catch (Exception error)
|
||
{
|
||
Console.WriteLine(
|
||
$"streaming: origin-recenter preparation will resume: {error}");
|
||
return false;
|
||
}
|
||
finally
|
||
{
|
||
_advancingOriginRecenter = false;
|
||
}
|
||
}
|
||
|
||
private void BeginFullWindowRetirement()
|
||
{
|
||
_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>
|
||
/// Adopts immutable worker results and advances explicit priority FIFOs
|
||
/// under the one typed frame meter. Destination and unload work win queue
|
||
/// order but no class bypasses time, bytes, entities, uploads, or retire
|
||
/// operation limits.
|
||
/// </summary>
|
||
private void DrainAndApply()
|
||
{
|
||
StreamingWorkMeter meter = _activeWorkMeter
|
||
?? throw new InvalidOperationException(
|
||
"Completion scheduling requires an active frame meter.");
|
||
|
||
AdmitCompletions(meter);
|
||
bool executed = false;
|
||
while (_completionQueue.TryPeekNext(
|
||
_isPublicationBlockedByRetirement,
|
||
out StreamingQueuedCompletion? completion))
|
||
{
|
||
StreamingQueuedCompletion work = completion
|
||
?? throw new InvalidOperationException(
|
||
"The completion queue returned a null head.");
|
||
try
|
||
{
|
||
using StreamingWorkMeter.LaneScope lane = meter.EnterLane(
|
||
work.Priority == StreamingCompletionPriority.Destination
|
||
&& work.RevealGeneration == ActiveRevealGeneration
|
||
? StreamingWorkLane.Destination
|
||
: StreamingWorkLane.NonDestination);
|
||
LandblockPublicationAdvance advance = ApplyResult(
|
||
work,
|
||
meter,
|
||
ensureProgress: !executed);
|
||
executed |= advance.Progressed;
|
||
if (!advance.Completed)
|
||
break;
|
||
_completionQueue.RemoveHead(work);
|
||
}
|
||
catch
|
||
{
|
||
// Concrete publication failures retain their exact owner
|
||
// receipt and queue head. A compatibility callback without a
|
||
// receipt cannot safely replay, so consume only that result.
|
||
if (!_presentation.HasPendingPublication(work.Result))
|
||
_completionQueue.RemoveHead(work);
|
||
throw;
|
||
}
|
||
}
|
||
}
|
||
|
||
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))
|
||
{
|
||
LandblockStreamResult result = peeked
|
||
?? throw new InvalidOperationException(
|
||
"The completion source returned a null peek.");
|
||
bool stale = IsStaleGeneration(result)
|
||
&& !_presentation.HasPendingPublication(result);
|
||
StreamingCompletionPriority priority = stale
|
||
? StreamingCompletionPriority.Control
|
||
: ClassifyCompletion(result);
|
||
LandblockStreamCostEstimate estimate =
|
||
LandblockStreamResultCost.Estimate(result);
|
||
StreamingWorkCost admissionCost = stale
|
||
? new StreamingWorkCost(CompletionAdmissions: 1)
|
||
: new StreamingWorkCost(
|
||
CompletionAdmissions:
|
||
estimate.Work.CompletionAdmissions,
|
||
AdoptedCpuBytes:
|
||
estimate.Work.AdoptedCpuBytes);
|
||
using StreamingWorkMeter.LaneScope lane = meter.EnterLane(
|
||
priority == StreamingCompletionPriority.Destination
|
||
? StreamingWorkLane.Destination
|
||
: StreamingWorkLane.NonDestination);
|
||
StreamingWorkAdmission admission = meter.TryReserve(
|
||
admissionCost,
|
||
"completion-admission");
|
||
if (admission == StreamingWorkAdmission.Yielded)
|
||
return;
|
||
|
||
try
|
||
{
|
||
if (!_completionSource.TryRead(
|
||
out LandblockStreamResult? consumed)
|
||
|| !ReferenceEquals(result, consumed))
|
||
{
|
||
throw new InvalidOperationException(
|
||
"The single-consumer completion source changed between peek and read.");
|
||
}
|
||
|
||
if (!stale)
|
||
{
|
||
_completionQueue.Enqueue(new StreamingQueuedCompletion(
|
||
result,
|
||
estimate,
|
||
priority,
|
||
priority == StreamingCompletionPriority.Destination
|
||
? ActiveRevealGeneration
|
||
: 0L,
|
||
result.Generation,
|
||
checked(_nextCompletionSequence++),
|
||
Stopwatch.GetTimestamp()));
|
||
}
|
||
meter.Complete();
|
||
}
|
||
catch
|
||
{
|
||
meter.Fail();
|
||
throw;
|
||
}
|
||
}
|
||
}
|
||
|
||
private StreamingCompletionPriority ClassifyCompletion(
|
||
LandblockStreamResult result)
|
||
{
|
||
if (result is LandblockStreamResult.Loaded
|
||
or LandblockStreamResult.Promoted
|
||
&& IsDestinationWork(result.LandblockId))
|
||
{
|
||
return StreamingCompletionPriority.Destination;
|
||
}
|
||
|
||
return result switch
|
||
{
|
||
LandblockStreamResult.Failed
|
||
or LandblockStreamResult.WorkerCrashed =>
|
||
StreamingCompletionPriority.Control,
|
||
LandblockStreamResult.Unloaded =>
|
||
StreamingCompletionPriority.Unload,
|
||
LandblockStreamResult.Promoted
|
||
or LandblockStreamResult.Loaded
|
||
{
|
||
Tier: LandblockStreamTier.Near,
|
||
} =>
|
||
StreamingCompletionPriority.Near,
|
||
_ => StreamingCompletionPriority.Far,
|
||
};
|
||
}
|
||
|
||
private static LandblockPublicationAdvance RunSimpleResultOperation(
|
||
StreamingWorkMeter meter,
|
||
string stage,
|
||
bool ensureProgress,
|
||
Action operation)
|
||
{
|
||
StreamingWorkAdmission admission = meter.TryReserve(
|
||
default,
|
||
stage,
|
||
ensureProgress);
|
||
if (admission == StreamingWorkAdmission.Yielded)
|
||
return new LandblockPublicationAdvance(false, false);
|
||
|
||
try
|
||
{
|
||
operation();
|
||
meter.Complete();
|
||
return new LandblockPublicationAdvance(true, true);
|
||
}
|
||
catch
|
||
{
|
||
meter.Fail();
|
||
throw;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Apply a single <see cref="LandblockStreamResult"/> with the full side-
|
||
/// effects: terrain upload, GPU state, and the re-hydration callback.
|
||
/// All priority queues route through this one publication path.
|
||
/// </summary>
|
||
private LandblockPublicationAdvance ApplyResult(
|
||
StreamingQueuedCompletion work,
|
||
StreamingWorkMeter meter,
|
||
bool ensureProgress)
|
||
{
|
||
LandblockStreamResult result = work.Result;
|
||
if (_presentation.HasPendingPublication(result))
|
||
{
|
||
LandblockPublicationAdvance resumed =
|
||
_presentation.ResumePublication(
|
||
result,
|
||
meter,
|
||
ensureProgress);
|
||
if (resumed.Completed)
|
||
ReconcileCompletedPendingPublication(result.LandblockId);
|
||
return resumed;
|
||
}
|
||
|
||
if (IsStaleGeneration(result))
|
||
{
|
||
// A worker can finish one old load/unload after a hard recenter.
|
||
// Landblock id membership cannot distinguish overlapping windows;
|
||
// generation is the logical streaming incarnation boundary.
|
||
return RunSimpleResultOperation(
|
||
meter,
|
||
"execute-stale-generation",
|
||
ensureProgress,
|
||
static () => { });
|
||
}
|
||
|
||
if (result is LandblockStreamResult.Unloaded
|
||
&& _region?.TryGetDesiredTier(result.LandblockId, out _) == true)
|
||
{
|
||
// Normal recenter does not advance the hard-window generation. A
|
||
// rapid away->back can therefore re-own an id while its same-
|
||
// generation unload is already in the worker outbox. Current
|
||
// region ownership wins; the old unload must not tear it down.
|
||
return RunSimpleResultOperation(
|
||
meter,
|
||
"execute-reowned-unload",
|
||
ensureProgress,
|
||
static () => { });
|
||
}
|
||
|
||
if (result is LandblockStreamResult.Loaded
|
||
or LandblockStreamResult.Promoted)
|
||
{
|
||
if (_region is null
|
||
|| !_region.TryGetDesiredTier(result.LandblockId, out var desiredTier))
|
||
{
|
||
// A hard recenter/collapse can leave one worker job already in
|
||
// flight. Its completion belongs to the old region and must not
|
||
// resurrect a landblock the new StreamingRegion never owns.
|
||
return RunSimpleResultOperation(
|
||
meter,
|
||
"execute-undesired-load",
|
||
ensureProgress,
|
||
static () => { });
|
||
}
|
||
|
||
bool isNearCompletion = result is LandblockStreamResult.Promoted
|
||
or LandblockStreamResult.Loaded { Tier: LandblockStreamTier.Near };
|
||
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
|
||
// supersedes Far/Unload work but intentionally does not dedupe
|
||
// high-priority jobs. Publication is idempotent at this owner:
|
||
// never append statics, replay defaults, or reapply the complete
|
||
// Near transaction to an already-Near landblock.
|
||
return RunSimpleResultOperation(
|
||
meter,
|
||
"execute-duplicate-near",
|
||
ensureProgress,
|
||
static () => { });
|
||
}
|
||
if (desiredTier == LandblockStreamTier.Far && isNearCompletion)
|
||
{
|
||
// Recenter can demote a Near job before its first completion.
|
||
// StreamingRegion already considers that landblock Far, so it
|
||
// deliberately emits no second LoadFar job. If no terrain is
|
||
// resident yet, publish the completed heightmap/mesh as a Far
|
||
// result while discarding its entity and EnvCell layers. This
|
||
// is the same payload a fresh LoadFar would have produced and
|
||
// closes the in-flight Near -> desired Far lifecycle without a
|
||
// duplicate worker read or a permanent terrain hole.
|
||
if (!_state.IsLoaded(result.LandblockId))
|
||
{
|
||
switch (result)
|
||
{
|
||
case LandblockStreamResult.Loaded loaded:
|
||
return _presentation.PublishAsFar(
|
||
loaded,
|
||
loaded.Build,
|
||
loaded.MeshData,
|
||
meter,
|
||
ensureProgress);
|
||
case LandblockStreamResult.Promoted promoted:
|
||
return _presentation.PublishAsFar(
|
||
promoted,
|
||
promoted.Build,
|
||
promoted.MeshData,
|
||
meter,
|
||
ensureProgress);
|
||
}
|
||
}
|
||
return RunSimpleResultOperation(
|
||
meter,
|
||
"execute-already-far",
|
||
ensureProgress,
|
||
static () => { });
|
||
}
|
||
}
|
||
|
||
switch (result)
|
||
{
|
||
case LandblockStreamResult.Loaded loaded:
|
||
if (loaded.Tier == LandblockStreamTier.Far
|
||
&& _state.IsNearTier(loaded.LandblockId))
|
||
{
|
||
// This Far completion was queued before a newer Near load
|
||
// or promotion. Applying it would erase the entity/cell
|
||
// layer that now owns the landblock.
|
||
return RunSimpleResultOperation(
|
||
meter,
|
||
"execute-stale-far-tier",
|
||
ensureProgress,
|
||
static () => { });
|
||
}
|
||
return _presentation.PublishLoaded(
|
||
loaded,
|
||
work.Estimate,
|
||
meter,
|
||
ensureProgress);
|
||
case LandblockStreamResult.Promoted promoted:
|
||
// PromoteToNear carries a complete build and mesh because the
|
||
// streamer deliberately lets it supersede a queued LoadFar. If
|
||
// 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.
|
||
return _presentation.PublishPromoted(
|
||
promoted,
|
||
mergeIntoExistingLandblock:
|
||
_state.IsLoaded(promoted.LandblockId),
|
||
work.Estimate,
|
||
meter,
|
||
ensureProgress);
|
||
case LandblockStreamResult.Unloaded unloaded:
|
||
return RunSimpleResultOperation(
|
||
meter,
|
||
"execute-unload",
|
||
ensureProgress,
|
||
() => _presentation.EnqueueFullRetirement(
|
||
unloaded.LandblockId));
|
||
case LandblockStreamResult.Failed failed:
|
||
return RunSimpleResultOperation(
|
||
meter,
|
||
"execute-load-failure",
|
||
ensureProgress,
|
||
() => Console.WriteLine(
|
||
$"streaming: load failed for 0x{failed.LandblockId:X8}: {failed.Error}"));
|
||
case LandblockStreamResult.WorkerCrashed crashed:
|
||
return RunSimpleResultOperation(
|
||
meter,
|
||
"execute-worker-crash",
|
||
ensureProgress,
|
||
() => Console.WriteLine(
|
||
$"streaming: worker CRASHED: {crashed.Error}"));
|
||
default:
|
||
throw new InvalidOperationException(
|
||
$"Unsupported streaming result {result.GetType().Name}.");
|
||
}
|
||
}
|
||
|
||
private void ReconcileCompletedPendingPublication(uint landblockId)
|
||
{
|
||
if (_region is null
|
||
|| !_region.TryGetDesiredTier(landblockId, out LandblockStreamTier desiredTier))
|
||
{
|
||
if (_state.IsLoaded(landblockId))
|
||
_presentation.EnqueueFullRetirement(landblockId);
|
||
return;
|
||
}
|
||
|
||
if (desiredTier == LandblockStreamTier.Far
|
||
&& _state.IsNearTier(landblockId))
|
||
{
|
||
_presentation.EnqueueNearLayerRetirement(landblockId);
|
||
}
|
||
}
|
||
|
||
private bool IsStaleGeneration(LandblockStreamResult result) =>
|
||
result is not LandblockStreamResult.WorkerCrashed
|
||
&& result.Generation != _generation;
|
||
|
||
private bool IsPublicationBlockedByRetirement(LandblockStreamResult result) =>
|
||
result is LandblockStreamResult.Loaded or LandblockStreamResult.Promoted
|
||
&& _presentation.IsRetirementPending(result.LandblockId);
|
||
|
||
/// <summary>
|
||
/// Returns the landblock id associated with <paramref name="result"/>.
|
||
/// For <see cref="LandblockStreamResult.WorkerCrashed"/> this is 0 by
|
||
/// convention (not tied to a specific landblock).
|
||
/// </summary>
|
||
private static uint ResultLandblockId(LandblockStreamResult result) => result.LandblockId;
|
||
}
|