perf(streaming): cursor publication across frame budgets
This commit is contained in:
Erik
parent
bb16f74fd4
commit
98f1ac8934
32 changed files with 2215 additions and 823 deletions
|
|
@ -370,7 +370,11 @@ public sealed class StreamingController : IStreamingFrameBackend
|
|||
// 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 (!ConvergePendingPublications())
|
||||
{
|
||||
_deferredRadiiRequest = (nearRadius, farRadius);
|
||||
return;
|
||||
}
|
||||
|
||||
if (_collapsed || _region is null)
|
||||
{
|
||||
|
|
@ -490,6 +494,12 @@ public sealed class StreamingController : IStreamingFrameBackend
|
|||
_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
|
||||
{
|
||||
|
|
@ -497,15 +507,14 @@ public sealed class StreamingController : IStreamingFrameBackend
|
|||
for (int i = 0; i < pending.Count; i++)
|
||||
{
|
||||
LandblockStreamResult result = pending[i];
|
||||
if (!TryObserveResultOperation(
|
||||
result,
|
||||
"publication-retry",
|
||||
() => _presentation.ResumePublication(result),
|
||||
ensureProgress: !progressed))
|
||||
{
|
||||
LandblockPublicationAdvance advance =
|
||||
_presentation.ResumePublication(
|
||||
result,
|
||||
_activeWorkMeter,
|
||||
ensureProgress: !progressed);
|
||||
progressed |= advance.Progressed;
|
||||
if (!advance.Completed)
|
||||
return false;
|
||||
}
|
||||
progressed = true;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
|
@ -1487,23 +1496,16 @@ public sealed class StreamingController : IStreamingFrameBackend
|
|||
StreamingQueuedCompletion work = completion
|
||||
?? throw new InvalidOperationException(
|
||||
"The completion queue returned a null head.");
|
||||
StreamingWorkCost executionCost =
|
||||
IsStaleGeneration(work.Result)
|
||||
? default
|
||||
: work.ExecutionCost;
|
||||
StreamingWorkAdmission admission = meter.TryReserve(
|
||||
executionCost,
|
||||
$"execute-{work.Priority}",
|
||||
ensureProgress: !executed);
|
||||
if (admission == StreamingWorkAdmission.Yielded)
|
||||
break;
|
||||
|
||||
try
|
||||
{
|
||||
ApplyResult(work.Result);
|
||||
LandblockPublicationAdvance advance = ApplyResult(
|
||||
work,
|
||||
meter,
|
||||
ensureProgress: !executed);
|
||||
executed |= advance.Progressed;
|
||||
if (!advance.Completed)
|
||||
break;
|
||||
_completionQueue.RemoveHead(work);
|
||||
meter.Complete();
|
||||
executed = true;
|
||||
}
|
||||
catch
|
||||
{
|
||||
|
|
@ -1512,7 +1514,6 @@ public sealed class StreamingController : IStreamingFrameBackend
|
|||
// receipt cannot safely replay, so consume only that result.
|
||||
if (!_presentation.HasPendingPublication(work.Result))
|
||||
_completionQueue.RemoveHead(work);
|
||||
meter.Fail();
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
|
@ -1625,39 +1626,24 @@ public sealed class StreamingController : IStreamingFrameBackend
|
|||
};
|
||||
}
|
||||
|
||||
private bool TryObserveResultOperation(
|
||||
LandblockStreamResult result,
|
||||
private static LandblockPublicationAdvance RunSimpleResultOperation(
|
||||
StreamingWorkMeter meter,
|
||||
string stage,
|
||||
Action operation,
|
||||
StreamingWorkCost? knownCost = null,
|
||||
bool ensureProgress = false)
|
||||
bool ensureProgress,
|
||||
Action operation)
|
||||
{
|
||||
StreamingWorkMeter? meter = _activeWorkMeter;
|
||||
if (meter is null)
|
||||
{
|
||||
operation();
|
||||
return true;
|
||||
}
|
||||
|
||||
StreamingWorkCost aggregate = knownCost
|
||||
?? LandblockStreamResultCost.Estimate(result).Work;
|
||||
StreamingWorkCost cost = new(
|
||||
EntityOperations: aggregate.EntityOperations,
|
||||
GpuUploadBytes: aggregate.GpuUploadBytes,
|
||||
GlRetireOperations:
|
||||
result is LandblockStreamResult.Unloaded ? 1 : 0);
|
||||
StreamingWorkAdmission admission = meter.TryReserve(
|
||||
cost,
|
||||
default,
|
||||
stage,
|
||||
ensureProgress);
|
||||
if (admission == StreamingWorkAdmission.Yielded)
|
||||
return false;
|
||||
return new LandblockPublicationAdvance(false, false);
|
||||
|
||||
try
|
||||
{
|
||||
operation();
|
||||
meter.Complete();
|
||||
return true;
|
||||
return new LandblockPublicationAdvance(true, true);
|
||||
}
|
||||
catch
|
||||
{
|
||||
|
|
@ -1671,13 +1657,22 @@ public sealed class StreamingController : IStreamingFrameBackend
|
|||
/// effects: terrain upload, GPU state, and the re-hydration callback.
|
||||
/// All priority queues route through this one publication path.
|
||||
/// </summary>
|
||||
private void ApplyResult(LandblockStreamResult result)
|
||||
private LandblockPublicationAdvance ApplyResult(
|
||||
StreamingQueuedCompletion work,
|
||||
StreamingWorkMeter meter,
|
||||
bool ensureProgress)
|
||||
{
|
||||
LandblockStreamResult result = work.Result;
|
||||
if (_presentation.HasPendingPublication(result))
|
||||
{
|
||||
_presentation.ResumePublication(result);
|
||||
ReconcileCompletedPendingPublication(result.LandblockId);
|
||||
return;
|
||||
LandblockPublicationAdvance resumed =
|
||||
_presentation.ResumePublication(
|
||||
result,
|
||||
meter,
|
||||
ensureProgress);
|
||||
if (resumed.Completed)
|
||||
ReconcileCompletedPendingPublication(result.LandblockId);
|
||||
return resumed;
|
||||
}
|
||||
|
||||
if (IsStaleGeneration(result))
|
||||
|
|
@ -1685,7 +1680,11 @@ public sealed class StreamingController : IStreamingFrameBackend
|
|||
// 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;
|
||||
return RunSimpleResultOperation(
|
||||
meter,
|
||||
"execute-stale-generation",
|
||||
ensureProgress,
|
||||
static () => { });
|
||||
}
|
||||
|
||||
if (result is LandblockStreamResult.Unloaded
|
||||
|
|
@ -1695,7 +1694,11 @@ public sealed class StreamingController : IStreamingFrameBackend
|
|||
// 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;
|
||||
return RunSimpleResultOperation(
|
||||
meter,
|
||||
"execute-reowned-unload",
|
||||
ensureProgress,
|
||||
static () => { });
|
||||
}
|
||||
|
||||
if (result is LandblockStreamResult.Loaded
|
||||
|
|
@ -1707,7 +1710,11 @@ public sealed class StreamingController : IStreamingFrameBackend
|
|||
// 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;
|
||||
return RunSimpleResultOperation(
|
||||
meter,
|
||||
"execute-undesired-load",
|
||||
ensureProgress,
|
||||
static () => { });
|
||||
}
|
||||
|
||||
bool isNearCompletion = result is LandblockStreamResult.Promoted
|
||||
|
|
@ -1722,7 +1729,11 @@ public sealed class StreamingController : IStreamingFrameBackend
|
|||
// 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;
|
||||
return RunSimpleResultOperation(
|
||||
meter,
|
||||
"execute-duplicate-near",
|
||||
ensureProgress,
|
||||
static () => { });
|
||||
}
|
||||
if (desiredTier == LandblockStreamTier.Far && isNearCompletion)
|
||||
{
|
||||
|
|
@ -1739,14 +1750,26 @@ public sealed class StreamingController : IStreamingFrameBackend
|
|||
switch (result)
|
||||
{
|
||||
case LandblockStreamResult.Loaded loaded:
|
||||
_presentation.PublishAsFar(loaded, loaded.Build, loaded.MeshData);
|
||||
break;
|
||||
return _presentation.PublishAsFar(
|
||||
loaded,
|
||||
loaded.Build,
|
||||
loaded.MeshData,
|
||||
meter,
|
||||
ensureProgress);
|
||||
case LandblockStreamResult.Promoted promoted:
|
||||
_presentation.PublishAsFar(promoted, promoted.Build, promoted.MeshData);
|
||||
break;
|
||||
return _presentation.PublishAsFar(
|
||||
promoted,
|
||||
promoted.Build,
|
||||
promoted.MeshData,
|
||||
meter,
|
||||
ensureProgress);
|
||||
}
|
||||
}
|
||||
return;
|
||||
return RunSimpleResultOperation(
|
||||
meter,
|
||||
"execute-already-far",
|
||||
ensureProgress,
|
||||
static () => { });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1759,31 +1782,54 @@ public sealed class StreamingController : IStreamingFrameBackend
|
|||
// 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.
|
||||
break;
|
||||
return RunSimpleResultOperation(
|
||||
meter,
|
||||
"execute-stale-far-tier",
|
||||
ensureProgress,
|
||||
static () => { });
|
||||
}
|
||||
_presentation.PublishLoaded(loaded);
|
||||
break;
|
||||
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.
|
||||
_presentation.PublishPromoted(
|
||||
return _presentation.PublishPromoted(
|
||||
promoted,
|
||||
mergeIntoExistingLandblock: _state.IsLoaded(promoted.LandblockId));
|
||||
break;
|
||||
mergeIntoExistingLandblock:
|
||||
_state.IsLoaded(promoted.LandblockId),
|
||||
work.Estimate,
|
||||
meter,
|
||||
ensureProgress);
|
||||
case LandblockStreamResult.Unloaded unloaded:
|
||||
_presentation.EnqueueFullRetirement(unloaded.LandblockId);
|
||||
break;
|
||||
return RunSimpleResultOperation(
|
||||
meter,
|
||||
"execute-unload",
|
||||
ensureProgress,
|
||||
() => _presentation.EnqueueFullRetirement(
|
||||
unloaded.LandblockId));
|
||||
case LandblockStreamResult.Failed failed:
|
||||
Console.WriteLine(
|
||||
$"streaming: load failed for 0x{failed.LandblockId:X8}: {failed.Error}");
|
||||
break;
|
||||
return RunSimpleResultOperation(
|
||||
meter,
|
||||
"execute-load-failure",
|
||||
ensureProgress,
|
||||
() => Console.WriteLine(
|
||||
$"streaming: load failed for 0x{failed.LandblockId:X8}: {failed.Error}"));
|
||||
case LandblockStreamResult.WorkerCrashed crashed:
|
||||
Console.WriteLine(
|
||||
$"streaming: worker CRASHED: {crashed.Error}");
|
||||
break;
|
||||
return RunSimpleResultOperation(
|
||||
meter,
|
||||
"execute-worker-crash",
|
||||
ensureProgress,
|
||||
() => Console.WriteLine(
|
||||
$"streaming: worker CRASHED: {crashed.Error}"));
|
||||
default:
|
||||
throw new InvalidOperationException(
|
||||
$"Unsupported streaming result {result.GetType().Name}.");
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue