525 lines
22 KiB
C#
525 lines
22 KiB
C#
using System;
|
||
using System.Collections.Generic;
|
||
using System.Threading;
|
||
using System.Threading.Channels;
|
||
using System.Threading.Tasks;
|
||
using AcDream.Core.World;
|
||
|
||
namespace AcDream.App.Streaming;
|
||
|
||
/// <summary>
|
||
/// Services landblock load/unload requests by invoking caller-supplied
|
||
/// factory delegates (the production instance wraps
|
||
/// <see cref="LandblockLoader.Load"/> for loading and
|
||
/// <see cref="AcDream.Core.Terrain.LandblockMesh.Build"/> for the terrain
|
||
/// mesh) and posting results to an outbox the render thread drains once
|
||
/// per OnUpdate.
|
||
///
|
||
/// <para>
|
||
/// <b>Thread model (Phase A.5 T11+):</b> <see cref="Start"/> spawns a
|
||
/// dedicated background worker thread. <see cref="EnqueueLoad"/> and
|
||
/// <see cref="EnqueueUnload"/> write non-blocking to the inbox
|
||
/// <see cref="Channel{T}"/>; the worker drains it and posts
|
||
/// <see cref="LandblockStreamResult"/> records to the outbox.
|
||
/// </para>
|
||
///
|
||
/// <para>
|
||
/// <b>DatCollection thread safety</b> is provided by the caller:
|
||
/// GameWindow's <c>_datLock</c> (Phase A.5 T10) serialises all
|
||
/// <c>DatCollection.Get<T></c> calls. Both factory closures passed at
|
||
/// construction acquire that lock before reading dats. The worker never
|
||
/// touches <c>DatCollection</c> directly — it only calls the factories.
|
||
/// </para>
|
||
///
|
||
/// <para>
|
||
/// Unloads pass through the outbox as <see cref="LandblockStreamResult.Unloaded"/>
|
||
/// records so the render thread can release GPU state on the next drain —
|
||
/// the streamer never touches GPU resources directly.
|
||
/// </para>
|
||
///
|
||
/// <remarks>
|
||
/// Threading: <see cref="DrainCompletions"/> must be called from a single
|
||
/// consumer thread (the render thread in production). All other public
|
||
/// methods are thread-safe.
|
||
/// </remarks>
|
||
/// </summary>
|
||
public sealed class LandblockStreamer : IDisposable
|
||
{
|
||
/// <summary>
|
||
/// Default drain batch size. Tuned to cap GPU upload work the render
|
||
/// thread does per frame while still draining a moderate backlog in a
|
||
/// few frames. Callers can override on a per-call basis.
|
||
/// </summary>
|
||
public const int DefaultDrainBatchSize = 4;
|
||
|
||
private readonly Func<LandblockBuildRequest, LandblockBuild?> _loadLandblock;
|
||
private readonly bool _supportsRequestOrigin;
|
||
private readonly Func<uint, LoadedLandblock?, AcDream.Core.Terrain.LandblockMeshData?> _buildMeshOrNull;
|
||
private readonly Channel<LandblockStreamJob> _inbox;
|
||
private readonly Channel<LandblockStreamResult> _outbox;
|
||
private readonly CancellationTokenSource _cancel = new();
|
||
private readonly object _inboxGate = new();
|
||
private Thread? _worker;
|
||
private Exception? _workerFailure;
|
||
private int _disposed;
|
||
private readonly object _disposeGate = new();
|
||
private bool _disposeCompleted;
|
||
|
||
/// <summary>
|
||
/// Primary constructor. The factory receives the complete immutable
|
||
/// request, including job kind and enqueue-time world origin. ISSUE #54
|
||
/// uses the kind to skip entity hydration for heightmap-only far loads.
|
||
/// </summary>
|
||
private LandblockStreamer(
|
||
Func<LandblockBuildRequest, LandblockBuild?> loadLandblock,
|
||
Func<uint, LoadedLandblock?, AcDream.Core.Terrain.LandblockMeshData?>? buildMeshOrNull,
|
||
bool supportsRequestOrigin)
|
||
{
|
||
_loadLandblock = loadLandblock;
|
||
_supportsRequestOrigin = supportsRequestOrigin;
|
||
// Default: no mesh build (returns null → Failed result). Production
|
||
// wires in LandblockMesh.Build via the T12 construction site.
|
||
_buildMeshOrNull = buildMeshOrNull ?? ((_, _) => null);
|
||
_inbox = Channel.CreateUnbounded<LandblockStreamJob>(
|
||
new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });
|
||
_outbox = Channel.CreateUnbounded<LandblockStreamResult>(
|
||
new UnboundedChannelOptions { SingleReader = true, SingleWriter = true });
|
||
}
|
||
|
||
/// <summary>
|
||
/// Creates the request-aware production streamer. A named factory keeps
|
||
/// legacy one-parameter loader lambdas source-compatible without an
|
||
/// ambiguous constructor overload.
|
||
/// </summary>
|
||
public static LandblockStreamer CreateForRequests(
|
||
Func<LandblockBuildRequest, LandblockBuild?> loadLandblock,
|
||
Func<uint, LoadedLandblock?, AcDream.Core.Terrain.LandblockMeshData?>? buildMeshOrNull = null) =>
|
||
new(loadLandblock, buildMeshOrNull, supportsRequestOrigin: true);
|
||
|
||
/// <summary>
|
||
/// Compatibility constructor for build factories that predate the
|
||
/// immutable request/origin seam.
|
||
/// </summary>
|
||
public LandblockStreamer(
|
||
Func<uint, LandblockStreamJobKind, LandblockBuild?> loadLandblock,
|
||
Func<uint, LoadedLandblock?, AcDream.Core.Terrain.LandblockMeshData?>? buildMeshOrNull = null)
|
||
: this(
|
||
request => loadLandblock(request.LandblockId, request.Kind) is { } build
|
||
? build
|
||
: null,
|
||
buildMeshOrNull,
|
||
supportsRequestOrigin: false)
|
||
{
|
||
}
|
||
|
||
/// <summary>
|
||
/// Compatibility constructor for loaders that have no App-owned cell payload.
|
||
/// Production uses the <see cref="LandblockBuild"/> overload so render state
|
||
/// remains attached to the exact streaming completion that produced it.
|
||
/// </summary>
|
||
public LandblockStreamer(
|
||
Func<uint, LandblockStreamJobKind, LoadedLandblock?> loadLandblock,
|
||
Func<uint, LoadedLandblock?, AcDream.Core.Terrain.LandblockMeshData?>? buildMeshOrNull = null)
|
||
: this(
|
||
request => loadLandblock(request.LandblockId, request.Kind) is { } landblock
|
||
? new LandblockBuild(landblock, Origin: request.Origin)
|
||
: null,
|
||
buildMeshOrNull,
|
||
supportsRequestOrigin: false)
|
||
{
|
||
}
|
||
|
||
/// <summary>
|
||
/// Back-compat overload — wraps a kind-agnostic factory so existing test code
|
||
/// that doesn't care about the JobKind branch keeps compiling. The wrapper
|
||
/// ignores the kind and calls the factory once per LB regardless of tier.
|
||
/// New production code should use <see cref="CreateForRequests"/>.
|
||
/// </summary>
|
||
public LandblockStreamer(
|
||
Func<uint, LoadedLandblock?> loadLandblock,
|
||
Func<uint, LoadedLandblock?, AcDream.Core.Terrain.LandblockMeshData?>? buildMeshOrNull = null)
|
||
: this(
|
||
request => loadLandblock(request.LandblockId) is { } landblock
|
||
? new LandblockBuild(landblock, Origin: request.Origin)
|
||
: null,
|
||
buildMeshOrNull,
|
||
supportsRequestOrigin: false)
|
||
{
|
||
}
|
||
|
||
/// <summary>
|
||
/// Activate the dedicated background worker thread. Idempotent and
|
||
/// thread-safe: concurrent callers will only spawn one worker; subsequent
|
||
/// calls are no-ops. Serialized with disposal so a worker can never start
|
||
/// after the owning DAT lifetime has been released.
|
||
/// </summary>
|
||
public void Start()
|
||
{
|
||
lock (_disposeGate)
|
||
{
|
||
if (System.Threading.Volatile.Read(ref _disposed) != 0)
|
||
throw new ObjectDisposedException(nameof(LandblockStreamer));
|
||
if (_worker is not null)
|
||
return;
|
||
|
||
var worker = new Thread(WorkerLoop)
|
||
{
|
||
IsBackground = true,
|
||
Name = "acdream.streaming.worker",
|
||
};
|
||
worker.Start();
|
||
_worker = worker;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Non-blocking enqueue. The worker drains the inbox and posts a
|
||
/// <see cref="LandblockStreamResult.Loaded"/> (or
|
||
/// <see cref="LandblockStreamResult.Failed"/>) to the outbox.
|
||
/// </summary>
|
||
public void EnqueueLoad(
|
||
uint landblockId,
|
||
LandblockStreamJobKind kind = LandblockStreamJobKind.LoadNear,
|
||
ulong generation = 0)
|
||
{
|
||
if (_supportsRequestOrigin)
|
||
{
|
||
throw new InvalidOperationException(
|
||
"Request-aware landblock loaders require an explicit captured origin.");
|
||
}
|
||
EnqueueLoad(new LandblockBuildRequest(
|
||
landblockId,
|
||
kind,
|
||
generation,
|
||
default));
|
||
}
|
||
|
||
/// <summary>
|
||
/// Non-blocking enqueue of the exact request captured by the update
|
||
/// thread. The worker and completion retain its origin unchanged.
|
||
/// </summary>
|
||
public void EnqueueLoad(LandblockBuildRequest request)
|
||
{
|
||
if (System.Threading.Volatile.Read(ref _disposed) != 0)
|
||
throw new ObjectDisposedException(nameof(LandblockStreamer));
|
||
if (_supportsRequestOrigin && !request.Origin.IsSpecified)
|
||
{
|
||
throw new InvalidOperationException(
|
||
"Request-aware landblock loaders require a specified captured origin.");
|
||
}
|
||
if (!_supportsRequestOrigin && request.Origin.IsSpecified)
|
||
{
|
||
throw new InvalidOperationException(
|
||
"This compatibility landblock loader cannot consume a non-default build origin. " +
|
||
"Use LandblockStreamer.CreateForRequests.");
|
||
}
|
||
AcDream.Core.Physics.PhysicsDiagnostics.LogTeleport(
|
||
"ENQ",
|
||
request.LandblockId,
|
||
$"kind={request.Kind} origin=({request.Origin.CenterX:X2},{request.Origin.CenterY:X2})");
|
||
WriteJob(new LandblockStreamJob.Load(
|
||
request.LandblockId,
|
||
request.Kind,
|
||
request.Generation,
|
||
request.Origin));
|
||
}
|
||
|
||
/// <summary>
|
||
/// Non-blocking enqueue. The worker posts a
|
||
/// <see cref="LandblockStreamResult.Unloaded"/> to the outbox.
|
||
/// </summary>
|
||
public void EnqueueUnload(uint landblockId, ulong generation = 0)
|
||
{
|
||
WriteJob(new LandblockStreamJob.Unload(landblockId, generation));
|
||
}
|
||
|
||
/// <summary>
|
||
/// Cancel every queued-but-not-started Load. Posts a
|
||
/// <see cref="LandblockStreamJob.ClearLoads"/> control job which the worker
|
||
/// honours at read time, dropping all pending Loads from both priority
|
||
/// queues (Unloads survive). Used on the dungeon-entry edge to abort the
|
||
/// in-flight 25×25 neighbor window so the ~129 ocean-grid dungeons never
|
||
/// finish loading (#133 FPS). Loads the worker has ALREADY dequeued still
|
||
/// complete; the StreamingController's collapsed-sweep unloads those few.
|
||
/// </summary>
|
||
public void ClearPendingLoads()
|
||
{
|
||
WriteJob(new LandblockStreamJob.ClearLoads());
|
||
}
|
||
|
||
private void WriteJob(LandblockStreamJob job)
|
||
{
|
||
// Serialize the writer's terminal transition with enqueue. Without
|
||
// this small lifecycle gate a worker crash could complete the channel
|
||
// between the caller's state check and TryWrite, silently dropping a
|
||
// landblock request. Enqueues are destination-boundary events, not a
|
||
// per-frame hot path.
|
||
lock (_inboxGate)
|
||
{
|
||
if (System.Threading.Volatile.Read(ref _disposed) != 0)
|
||
throw new ObjectDisposedException(nameof(LandblockStreamer));
|
||
if (_workerFailure is { } failure)
|
||
throw new InvalidOperationException("The landblock streaming worker has terminated.", failure);
|
||
if (!_inbox.Writer.TryWrite(job))
|
||
throw new InvalidOperationException("The landblock streaming inbox is no longer accepting work.");
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Drain up to <paramref name="maxBatchSize"/> completed results.
|
||
/// Non-blocking. Call from the render thread once per OnUpdate.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// Must be called from a single consumer thread. The outbox channel is
|
||
/// configured with SingleReader = true and will throw on concurrent reads.
|
||
/// </remarks>
|
||
public IReadOnlyList<LandblockStreamResult> DrainCompletions(int maxBatchSize = DefaultDrainBatchSize)
|
||
{
|
||
var batch = new List<LandblockStreamResult>(maxBatchSize);
|
||
while (batch.Count < maxBatchSize && _outbox.Reader.TryRead(out var result))
|
||
batch.Add(result);
|
||
return batch;
|
||
}
|
||
|
||
private void WorkerLoop()
|
||
{
|
||
var highPriority = new Queue<LandblockStreamJob>();
|
||
var lowPriority = new Queue<LandblockStreamJob>();
|
||
|
||
try
|
||
{
|
||
// Safe to block: this is a dedicated worker thread with no
|
||
// SynchronizationContext, so .Result/.GetResult cannot deadlock
|
||
// against any captured continuation. Using the sync pattern
|
||
// here keeps the loop linear; an async-enumerable alternative
|
||
// would force WorkerLoop to be async Task and lose the
|
||
// simple thread-start shape.
|
||
while (!_cancel.Token.IsCancellationRequested)
|
||
{
|
||
if (highPriority.Count == 0 &&
|
||
lowPriority.Count == 0 &&
|
||
!_inbox.Reader.WaitToReadAsync(_cancel.Token).AsTask().GetAwaiter().GetResult())
|
||
{
|
||
break;
|
||
}
|
||
|
||
while (_inbox.Reader.TryRead(out var job))
|
||
{
|
||
if (job is LandblockStreamJob.ClearLoads)
|
||
{
|
||
// Dungeon-entry cancellation: drop every queued Load,
|
||
// keep Unloads. Handled at read time so it supersedes
|
||
// Loads sitting in the priority queues ahead of it.
|
||
DropLoadJobs(highPriority);
|
||
DropLoadJobs(lowPriority);
|
||
continue;
|
||
}
|
||
EnqueuePrioritized(job, highPriority, lowPriority);
|
||
}
|
||
|
||
if (highPriority.Count == 0 && lowPriority.Count == 0)
|
||
continue;
|
||
|
||
if (_cancel.Token.IsCancellationRequested) return;
|
||
var next = highPriority.Count > 0
|
||
? highPriority.Dequeue()
|
||
: lowPriority.Dequeue();
|
||
HandleJob(next);
|
||
}
|
||
}
|
||
catch (OperationCanceledException) { /* graceful shutdown */ }
|
||
catch (Exception ex)
|
||
{
|
||
// Last-ditch: surface via outbox so the caller at least sees
|
||
// something. We never retry a crashed worker.
|
||
lock (_inboxGate)
|
||
{
|
||
_workerFailure = ex;
|
||
_inbox.Writer.TryComplete(ex);
|
||
}
|
||
_outbox.Writer.TryWrite(new LandblockStreamResult.WorkerCrashed(ex.ToString()));
|
||
}
|
||
finally
|
||
{
|
||
_outbox.Writer.TryComplete();
|
||
}
|
||
}
|
||
|
||
private static void EnqueuePrioritized(
|
||
LandblockStreamJob job,
|
||
Queue<LandblockStreamJob> highPriority,
|
||
Queue<LandblockStreamJob> lowPriority)
|
||
{
|
||
if (job is LandblockStreamJob.Load
|
||
{
|
||
Kind: LandblockStreamJobKind.LoadNear or LandblockStreamJobKind.PromoteToNear
|
||
} high)
|
||
{
|
||
// Near-tier jobs are visible-content critical. They supersede an
|
||
// older queued LoadFar for the same landblock: LoadNear obviously
|
||
// loads everything, and PromoteToNear now carries mesh data so the
|
||
// render thread can run the full near-tier apply side effects. If a
|
||
// LoadFar is already being processed, the single worker naturally
|
||
// finishes it before the promotion is dequeued.
|
||
RemoveLowPriorityJobsForLandblock(
|
||
lowPriority,
|
||
high.LandblockId,
|
||
removeLoadFar: true,
|
||
removeUnload: true);
|
||
highPriority.Enqueue(job);
|
||
return;
|
||
}
|
||
|
||
lowPriority.Enqueue(job);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Drop every <see cref="LandblockStreamJob.Load"/> from a priority queue,
|
||
/// preserving Unloads (and any other control jobs). Rotates the queue once
|
||
/// in place. Used by the <see cref="LandblockStreamJob.ClearLoads"/> path.
|
||
/// </summary>
|
||
private static void DropLoadJobs(Queue<LandblockStreamJob> queue)
|
||
{
|
||
int count = queue.Count;
|
||
for (int i = 0; i < count; i++)
|
||
{
|
||
var job = queue.Dequeue();
|
||
if (job is not LandblockStreamJob.Load)
|
||
queue.Enqueue(job);
|
||
}
|
||
}
|
||
|
||
private static void RemoveLowPriorityJobsForLandblock(
|
||
Queue<LandblockStreamJob> queue,
|
||
uint landblockId,
|
||
bool removeLoadFar,
|
||
bool removeUnload)
|
||
{
|
||
int count = queue.Count;
|
||
for (int i = 0; i < count; i++)
|
||
{
|
||
var job = queue.Dequeue();
|
||
bool remove = job.LandblockId == landblockId && job switch
|
||
{
|
||
LandblockStreamJob.Load { Kind: LandblockStreamJobKind.LoadFar } => removeLoadFar,
|
||
LandblockStreamJob.Unload => removeUnload,
|
||
_ => false
|
||
};
|
||
if (!remove)
|
||
queue.Enqueue(job);
|
||
}
|
||
}
|
||
|
||
private void HandleJob(LandblockStreamJob job)
|
||
{
|
||
switch (job)
|
||
{
|
||
case LandblockStreamJob.Load load:
|
||
// ISSUE #54 (post-A.5): JobKind is now plumbed through to the
|
||
// factory, so far-tier loads can skip LandBlockInfo + scenery
|
||
// + interior hydration on the worker thread (heightmap-only).
|
||
// The post-load entity-strip below is retained as a Debug
|
||
// assertion + Release safety net for the case where a buggy
|
||
// factory returns far-tier with entities anyway.
|
||
try
|
||
{
|
||
var build = _loadLandblock(load.Request);
|
||
if (build is null)
|
||
{
|
||
_outbox.Writer.TryWrite(new LandblockStreamResult.Failed(
|
||
load.LandblockId, "LandblockLoader.Load returned null", load.Generation));
|
||
break;
|
||
}
|
||
if (build.Origin != load.Origin)
|
||
{
|
||
_outbox.Writer.TryWrite(new LandblockStreamResult.Failed(
|
||
load.LandblockId,
|
||
$"Landblock build origin {build.Origin} did not match request origin {load.Origin}",
|
||
load.Generation));
|
||
break;
|
||
}
|
||
var lb = build.Landblock;
|
||
if (load.Kind == LandblockStreamJobKind.PromoteToNear)
|
||
{
|
||
var promotedMesh = _buildMeshOrNull(load.LandblockId, lb);
|
||
if (promotedMesh is null)
|
||
{
|
||
_outbox.Writer.TryWrite(new LandblockStreamResult.Failed(
|
||
load.LandblockId, "buildMeshOrNull returned null", load.Generation));
|
||
break;
|
||
}
|
||
_outbox.Writer.TryWrite(new LandblockStreamResult.Promoted(
|
||
load.LandblockId, build, promotedMesh, load.Generation));
|
||
break;
|
||
}
|
||
var mesh = _buildMeshOrNull(load.LandblockId, lb);
|
||
if (mesh is null)
|
||
{
|
||
_outbox.Writer.TryWrite(new LandblockStreamResult.Failed(
|
||
load.LandblockId, "buildMeshOrNull returned null", load.Generation));
|
||
break;
|
||
}
|
||
var tier = load.Kind == LandblockStreamJobKind.LoadFar
|
||
? LandblockStreamTier.Far : LandblockStreamTier.Near;
|
||
if (tier == LandblockStreamTier.Far)
|
||
{
|
||
// Belt-and-suspenders: factory should have skipped
|
||
// entity hydration for LoadFar. If it didn't, fail
|
||
// loud in Debug builds and strip in Release.
|
||
bool hasNearPayload =
|
||
lb.Entities.Count > 0 ||
|
||
build.EnvCells is not null ||
|
||
lb.PhysicsDats is { } physicsDats &&
|
||
(physicsDats.Info is not null ||
|
||
physicsDats.EnvCells.Count > 0 ||
|
||
physicsDats.Environments.Count > 0 ||
|
||
physicsDats.Setups.Count > 0 ||
|
||
physicsDats.GfxObjs.Count > 0);
|
||
System.Diagnostics.Debug.Assert(
|
||
!hasNearPayload,
|
||
$"Far-tier factory returned Near payload for LB 0x{load.LandblockId:X8}");
|
||
lb = new LoadedLandblock(
|
||
lb.LandblockId,
|
||
lb.Heightmap,
|
||
System.Array.Empty<AcDream.Core.World.WorldEntity>(),
|
||
PhysicsDatBundle.Empty);
|
||
build = new LandblockBuild(lb, Origin: build.Origin);
|
||
}
|
||
_outbox.Writer.TryWrite(new LandblockStreamResult.Loaded(
|
||
load.LandblockId, tier, build, mesh, load.Generation));
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_outbox.Writer.TryWrite(new LandblockStreamResult.Failed(
|
||
load.LandblockId, ex.ToString(), load.Generation));
|
||
}
|
||
break;
|
||
|
||
case LandblockStreamJob.Unload unload:
|
||
_outbox.Writer.TryWrite(new LandblockStreamResult.Unloaded(
|
||
unload.LandblockId,
|
||
unload.Generation));
|
||
break;
|
||
}
|
||
}
|
||
|
||
public void Dispose()
|
||
{
|
||
lock (_disposeGate)
|
||
{
|
||
if (_disposeCompleted)
|
||
return;
|
||
|
||
System.Threading.Interlocked.Exchange(ref _disposed, 1);
|
||
_cancel.Cancel();
|
||
lock (_inboxGate)
|
||
_inbox.Writer.TryComplete();
|
||
// The owner releases the memory-mapped DAT immediately after this
|
||
// object. Join the actual worker without a grace-period timeout so
|
||
// no native read can survive into that teardown.
|
||
_worker?.Join();
|
||
_cancel.Dispose();
|
||
_disposeCompleted = true;
|
||
}
|
||
}
|
||
}
|