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;
///
/// Services landblock load/unload requests by invoking caller-supplied
/// factory delegates (the production instance wraps
/// for loading and
/// for the terrain
/// mesh) and posting results to an outbox the render thread drains once
/// per OnUpdate.
///
///
/// Thread model (#418): spawns a small pool of
/// dedicated background worker threads (default
/// ; the single-worker degenerate case is
/// the pre-#418 Phase A.5 T11+ shape). Jobs are striped across per-worker
/// lanes by landblock id, so every job for one landblock id executes on
/// one lane in enqueue order — a Load and Unload for the same id can never
/// race on two workers. and
/// write non-blocking to the owning lane's
/// inbox ; each worker drains its lane and posts
/// records to the shared outbox.
///
///
///
/// DatCollection thread safety is provided by the caller:
/// GameWindow's _datLock (Phase A.5 T10) serialises all
/// DatCollection.Get<T> calls. Both factory closures passed at
/// construction acquire that lock before reading dats. The workers never
/// touch DatCollection directly — they only call the factories.
///
///
///
/// Unloads pass through the outbox as
/// records so the render thread can release GPU state on the next drain —
/// the streamer never touches GPU resources directly.
///
///
///
/// Threading: must be called from a single
/// consumer thread (the render thread in production). All other public
/// methods are thread-safe. Completion arrival order is preserved per
/// landblock id (per lane); across different landblocks it is arbitrary,
/// which the single consumer already tolerates — the StreamingController's
/// admission classifies each result independently into per-priority FIFOs.
///
///
public sealed class LandblockStreamer : IDisposable, ILandblockCompletionSource
{
///
/// 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.
///
public const int DefaultDrainBatchSize = 4;
///
/// Default build worker pool size: leave two cores for the render and
/// update threads, cap at 8 (the login window's ~625 builds saturate
/// well before that), floor 1 (the pre-#418 single-worker shape).
///
public static int DefaultWorkerCount =>
Math.Max(1, Math.Min(Environment.ProcessorCount - 2, 8));
private readonly Func _loadLandblock;
private readonly bool _supportsRequestOrigin;
private readonly Func _buildMeshOrNull;
private readonly Channel[] _lanes;
private readonly Channel _outbox;
private readonly CancellationTokenSource _cancel = new();
private readonly object _inboxGate = new();
private Thread[]? _workers;
private int _activeWorkers;
private Exception? _workerFailure;
private int _completionBacklog;
private int _disposed;
private readonly object _disposeGate = new();
private bool _disposeCompleted;
///
/// 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.
///
private LandblockStreamer(
Func loadLandblock,
Func? buildMeshOrNull,
bool supportsRequestOrigin,
int? workerCount)
{
if (workerCount is < 1)
{
throw new ArgumentOutOfRangeException(
nameof(workerCount),
workerCount,
"The landblock build pool needs at least one worker.");
}
_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);
int lanes = workerCount ?? DefaultWorkerCount;
_lanes = new Channel[lanes];
for (int i = 0; i < lanes; i++)
{
_lanes[i] = Channel.CreateUnbounded(
new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });
}
// SingleWriter = false: every pool worker posts completions.
_outbox = Channel.CreateUnbounded(
new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });
}
///
/// Creates the request-aware production streamer. A named factory keeps
/// legacy one-parameter loader lambdas source-compatible without an
/// ambiguous constructor overload.
///
public static LandblockStreamer CreateForRequests(
Func loadLandblock,
Func? buildMeshOrNull = null,
int? workerCount = null) =>
new(loadLandblock, buildMeshOrNull, supportsRequestOrigin: true, workerCount);
///
/// Compatibility constructor for build factories that predate the
/// immutable request/origin seam.
///
public LandblockStreamer(
Func loadLandblock,
Func? buildMeshOrNull = null,
int? workerCount = null)
: this(
request => loadLandblock(request.LandblockId, request.Kind) is { } build
? build
: null,
buildMeshOrNull,
supportsRequestOrigin: false,
workerCount)
{
}
///
/// Compatibility constructor for loaders that have no App-owned cell payload.
/// Production uses the overload so render state
/// remains attached to the exact streaming completion that produced it.
///
public LandblockStreamer(
Func loadLandblock,
Func? buildMeshOrNull = null,
int? workerCount = null)
: this(
request => loadLandblock(request.LandblockId, request.Kind) is { } landblock
? new LandblockBuild(landblock, Origin: request.Origin)
: null,
buildMeshOrNull,
supportsRequestOrigin: false,
workerCount)
{
}
///
/// 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 .
///
public LandblockStreamer(
Func loadLandblock,
Func? buildMeshOrNull = null,
int? workerCount = null)
: this(
request => loadLandblock(request.LandblockId) is { } landblock
? new LandblockBuild(landblock, Origin: request.Origin)
: null,
buildMeshOrNull,
supportsRequestOrigin: false,
workerCount)
{
}
/// Configured pool size (also the lane count).
internal int WorkerCount => _lanes.Length;
///
/// Lane (worker) that owns every job for .
/// Landblock ids carry their identity in the high 16 bits (0xXXYYFFFF —
/// the low word is constant across ids), so the id is mixed with a
/// Knuth multiplicative hash before reduction; a bare modulo would map
/// every id to one lane. Internal so tests can construct deterministic
/// per-lane contention.
///
internal int LaneFor(uint landblockId)
{
if (_lanes.Length == 1)
return 0;
uint mixed = (landblockId >> 16) * 2654435761u;
return (int)(mixed % (uint)_lanes.Length);
}
///
/// Activate the dedicated background worker threads. Idempotent and
/// thread-safe: concurrent callers will only spawn one pool; subsequent
/// calls are no-ops. Serialized with disposal so a worker can never start
/// after the owning DAT lifetime has been released.
///
public void Start()
{
lock (_disposeGate)
{
if (System.Threading.Volatile.Read(ref _disposed) != 0)
throw new ObjectDisposedException(nameof(LandblockStreamer));
if (_workers is not null)
return;
var workers = new Thread[_lanes.Length];
// Armed before any worker can run so the last worker to exit —
// however fast — is the one that completes the outbox.
System.Threading.Volatile.Write(ref _activeWorkers, workers.Length);
for (int i = 0; i < workers.Length; i++)
{
int lane = i;
workers[i] = new Thread(() => WorkerLoop(lane))
{
IsBackground = true,
Name = workers.Length == 1
? "acdream.streaming.worker"
: $"acdream.streaming.worker.{lane}",
};
}
foreach (Thread worker in workers)
worker.Start();
_workers = workers;
}
}
///
/// Non-blocking enqueue. The worker drains the inbox and posts a
/// (or
/// ) to the outbox.
///
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));
}
///
/// Non-blocking enqueue of the exact request captured by the update
/// thread. The worker and completion retain its origin unchanged.
///
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));
}
///
/// Non-blocking enqueue. The worker posts a
/// to the outbox.
///
public void EnqueueUnload(uint landblockId, ulong generation = 0)
{
WriteJob(new LandblockStreamJob.Unload(landblockId, generation));
}
///
/// Cancel every queued-but-not-started Load. Posts a
/// control job to EVERY lane
/// (one atomic broadcast under the enqueue gate, so a load enqueued
/// before this call is ordered ahead of its lane's ClearLoads copy and
/// dropped) which each worker honours at read time, dropping all pending
/// Loads from both of its 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 a
/// worker has ALREADY dequeued still complete (up to one per worker);
/// the StreamingController's collapsed-sweep unloads those few.
///
public void ClearPendingLoads()
{
WriteJob(new LandblockStreamJob.ClearLoads());
}
private void WriteJob(LandblockStreamJob job)
{
// Serialize the writers' terminal transition with enqueue. Without
// this small lifecycle gate a worker crash could complete a lane
// between the caller's state check and TryWrite, silently dropping a
// landblock request. The same gate makes the ClearLoads broadcast
// atomic with respect to concurrent enqueues. 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("A landblock streaming worker has terminated.", failure);
if (job is LandblockStreamJob.ClearLoads)
{
foreach (Channel lane in _lanes)
{
if (!lane.Writer.TryWrite(job))
throw new InvalidOperationException("The landblock streaming inbox is no longer accepting work.");
}
return;
}
if (!_lanes[LaneFor(job.LandblockId)].Writer.TryWrite(job))
throw new InvalidOperationException("The landblock streaming inbox is no longer accepting work.");
}
}
///
/// Drain up to completed results.
/// Non-blocking. Call from the render thread once per OnUpdate.
///
///
/// Must be called from a single consumer thread. The outbox channel is
/// configured with SingleReader = true and will throw on concurrent reads.
///
public IReadOnlyList DrainCompletions(int maxBatchSize = DefaultDrainBatchSize)
{
var batch = new List(maxBatchSize);
while (batch.Count < maxBatchSize && TryRead(out var result))
{
if (result is null)
throw new InvalidOperationException(
"The completion channel returned a null result.");
batch.Add(result);
}
return batch;
}
public int BacklogCount => Math.Max(
0,
System.Threading.Volatile.Read(ref _completionBacklog));
public bool TryPeek(out LandblockStreamResult? result) =>
_outbox.Reader.TryPeek(out result);
public bool TryRead(out LandblockStreamResult? result)
{
if (!_outbox.Reader.TryRead(out result))
return false;
System.Threading.Interlocked.Decrement(ref _completionBacklog);
return true;
}
private void PublishResult(LandblockStreamResult result)
{
if (_outbox.Writer.TryWrite(result))
System.Threading.Interlocked.Increment(ref _completionBacklog);
}
private void WorkerLoop(int laneIndex)
{
ChannelReader inbox = _lanes[laneIndex].Reader;
var highPriority = new Queue();
var lowPriority = new Queue();
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.WaitToReadAsync(_cancel.Token).AsTask().GetAwaiter().GetResult())
{
break;
}
while (inbox.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; any worker crash
// terminates the pool (matching the single-worker contract that
// a crash ends all job processing). A sibling that merely
// observed the crashed worker's lane completion (the wrapped
// ChannelClosedException) exits quietly instead of reporting a
// second, spurious crash.
bool cascade;
lock (_inboxGate)
{
cascade = _workerFailure is not null && ex is ChannelClosedException;
_workerFailure ??= ex;
foreach (Channel lane in _lanes)
lane.Writer.TryComplete(ex);
}
if (!cascade)
{
PublishResult(new LandblockStreamResult.WorkerCrashed(
_lanes.Length == 1
? ex.ToString()
: $"worker {laneIndex}: {ex}"));
// Stop the sibling workers. Safe against Dispose: its
// CTS disposal only happens after every worker (this one
// included) has been joined.
_cancel.Cancel();
}
}
finally
{
// The outbox has N writers; only the last worker out may
// complete it, or a crashed worker would silently drop the
// still-running workers' completions.
if (Interlocked.Decrement(ref _activeWorkers) == 0)
_outbox.Writer.TryComplete();
}
}
private static void EnqueuePrioritized(
LandblockStreamJob job,
Queue highPriority,
Queue 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 owning lane's worker
// naturally finishes it before the promotion is dequeued (every
// job for one landblock id lives on one lane).
RemoveLowPriorityJobsForLandblock(
lowPriority,
high.LandblockId,
removeLoadFar: true,
removeUnload: true);
highPriority.Enqueue(job);
return;
}
lowPriority.Enqueue(job);
}
///
/// Drop every from a priority queue,
/// preserving Unloads (and any other control jobs). Rotates the queue once
/// in place. Used by the path.
///
private static void DropLoadJobs(Queue 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 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)
{
PublishResult(new LandblockStreamResult.Failed(
load.LandblockId, "LandblockLoader.Load returned null", load.Generation));
break;
}
if (build.Origin != load.Origin)
{
PublishResult(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)
{
PublishResult(new LandblockStreamResult.Failed(
load.LandblockId, "buildMeshOrNull returned null", load.Generation));
break;
}
PublishResult(new LandblockStreamResult.Promoted(
load.LandblockId, build, promotedMesh, load.Generation));
break;
}
var mesh = _buildMeshOrNull(load.LandblockId, lb);
if (mesh is null)
{
PublishResult(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(),
PhysicsDatBundle.Empty);
build = new LandblockBuild(lb, Origin: build.Origin);
}
PublishResult(new LandblockStreamResult.Loaded(
load.LandblockId, tier, build, mesh, load.Generation));
}
catch (Exception ex)
{
PublishResult(new LandblockStreamResult.Failed(
load.LandblockId, ex.ToString(), load.Generation));
}
break;
case LandblockStreamJob.Unload unload:
PublishResult(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)
{
foreach (Channel lane in _lanes)
lane.Writer.TryComplete();
}
// The owner releases the memory-mapped DAT immediately after this
// object. Join every actual worker without a grace-period timeout
// so no native read can survive into that teardown.
if (_workers is { } workers)
{
foreach (Thread worker in workers)
worker.Join();
}
_cancel.Dispose();
_disposeCompleted = true;
}
}
}